perxis.collections.helpers

 1from perxis.collections import collections_pb2
 2from dataclasses import dataclass, asdict
 3from typing_extensions import deprecated
 4
 5
 6@deprecated("This function is deprecated. Use `init_collections` instead.")
 7def make_collection_instances(
 8    schemes_dir: str, schemes_mapping: dict[str, str]
 9) -> list[collections_pb2.Collection]:
10    collections = []
11
12    for collection_id, collection_name in schemes_mapping.items():
13        with open(f"{schemes_dir}/{collection_id}.json", "r") as file:
14            collection_schema = file.read()
15
16        collection = collections_pb2.Collection(
17            id=collection_id,
18            name=collection_name,
19            schema=collection_schema
20        )
21        collections.append(collection)
22    return collections
23
24
25@dataclass
26class CollectionProps:
27   single: bool = False
28   system: bool = False
29   no_data: bool = False
30   hidden: bool = False
31   no_revisions: bool = False
32   no_archive: bool = False
33   no_publish: bool = False
34
35   def to_dict(self) -> dict:
36       return asdict(self)  # noqa
37
38
39def init_collections(
40    schemes_dir: str,
41    schemes_mapping: dict[str, str],
42    collections_settings_mapping: dict[str, CollectionProps],
43) -> list[collections_pb2.Collection]:
44
45    collections = []
46    for collection_id, collection_name in schemes_mapping.items():
47        with open(f"{schemes_dir}/{collection_id}.json", "r") as file:
48            collection_schema = file.read()
49
50        props = collections_settings_mapping.get(collection_id) or CollectionProps()
51
52        if not isinstance(props, CollectionProps):
53            raise TypeError("collections_settings_mapping value must be an instance of CollectionProps")
54
55        kwargs = {
56            **props.to_dict(),
57            "id": collection_id,
58            "name": collection_name,
59            "schema": collection_schema,
60        }
61        collection = collections_pb2.Collection(**kwargs)
62        collections.append(collection)
63    return collections
@deprecated('This function is deprecated. Use `init_collections` instead.')
def make_collection_instances( schemes_dir: str, schemes_mapping: dict[str, str]) -> list[collections.collections_pb2.Collection]:
 7@deprecated("This function is deprecated. Use `init_collections` instead.")
 8def make_collection_instances(
 9    schemes_dir: str, schemes_mapping: dict[str, str]
10) -> list[collections_pb2.Collection]:
11    collections = []
12
13    for collection_id, collection_name in schemes_mapping.items():
14        with open(f"{schemes_dir}/{collection_id}.json", "r") as file:
15            collection_schema = file.read()
16
17        collection = collections_pb2.Collection(
18            id=collection_id,
19            name=collection_name,
20            schema=collection_schema
21        )
22        collections.append(collection)
23    return collections
@dataclass
class CollectionProps:
26@dataclass
27class CollectionProps:
28   single: bool = False
29   system: bool = False
30   no_data: bool = False
31   hidden: bool = False
32   no_revisions: bool = False
33   no_archive: bool = False
34   no_publish: bool = False
35
36   def to_dict(self) -> dict:
37       return asdict(self)  # noqa
CollectionProps( single: bool = False, system: bool = False, no_data: bool = False, hidden: bool = False, no_revisions: bool = False, no_archive: bool = False, no_publish: bool = False)
single: bool = False
system: bool = False
no_data: bool = False
hidden: bool = False
no_revisions: bool = False
no_archive: bool = False
no_publish: bool = False
def to_dict(self) -> dict:
36   def to_dict(self) -> dict:
37       return asdict(self)  # noqa
def init_collections( schemes_dir: str, schemes_mapping: dict[str, str], collections_settings_mapping: dict[str, CollectionProps]) -> list[collections.collections_pb2.Collection]:
40def init_collections(
41    schemes_dir: str,
42    schemes_mapping: dict[str, str],
43    collections_settings_mapping: dict[str, CollectionProps],
44) -> list[collections_pb2.Collection]:
45
46    collections = []
47    for collection_id, collection_name in schemes_mapping.items():
48        with open(f"{schemes_dir}/{collection_id}.json", "r") as file:
49            collection_schema = file.read()
50
51        props = collections_settings_mapping.get(collection_id) or CollectionProps()
52
53        if not isinstance(props, CollectionProps):
54            raise TypeError("collections_settings_mapping value must be an instance of CollectionProps")
55
56        kwargs = {
57            **props.to_dict(),
58            "id": collection_id,
59            "name": collection_name,
60            "schema": collection_schema,
61        }
62        collection = collections_pb2.Collection(**kwargs)
63        collections.append(collection)
64    return collections