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
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