perxis.extensions.utils
Модуль содержит вспомогательные функции для сервисов расширений
1""" 2Модуль содержит вспомогательные функции для сервисов расширений 3""" 4 5from typing import Optional 6 7from perxis.extensions import manager_service_pb2 8from perxis.extensions.item_models import DataSourceItem, DataSourceData, SyncPolicyItem, SyncPolicyData 9 10 11def datasource_items_from_collections( 12 collections_map: dict[str, str], 13 datasource_data: DataSourceData, 14 override_for_collections: dict[str, DataSourceData] | None = None, 15 with_update: bool = True, 16 with_delete: bool = True, 17) -> list[DataSourceItem]: 18 """ 19 Создание записей источников данных на базе маппинга коллекций. 20 21 Arguments: 22 collections_map: Dict вида [collection_id, collection_name] 23 datasource_data: Dict со структурой DataSourceData для item'ов коллекций 24 override_for_collections: Dict для переопределения значений для отдельных коллекций 25 вида [collection_id, DataSourceData] 26 with_update: Флаг для возможности обновления существующих записей расширением. Может быть переопределён 27 в override_for_collections 28 with_delete: Флаг для возможности удаления существующих записей расширением. Может быть переопределён 29 в override_for_collections 30 Returns: 31 list[DataSourceItem] 32 """ 33 34 if not isinstance(override_for_collections, dict): 35 override_for_collections = {} 36 37 items = [] 38 39 for collection_id in collections_map: 40 if not collection_id: 41 continue 42 43 kwargs = { 44 "with_update": with_update, 45 "with_delete": with_delete, 46 **override_for_collections.get(collection_id, datasource_data), 47 "collection_id": collection_id, 48 } 49 50 items.append(DataSourceItem(**kwargs)) 51 52 return items 53 54 55def sync_policies_from_collections( 56 collections_map: dict[str, str], 57 sync_policies_data: SyncPolicyData, 58 override_for_collections: dict[str, SyncPolicyData] | None = None, 59 with_update: bool = True, 60 with_delete: bool = True, 61) -> list[SyncPolicyItem]: 62 """ 63 Создание записей синхронизации коллекций на базе маппинга коллекций 64 65 Arguments: 66 collections_map: Dict вида [collection_id, collection_name] 67 sync_policies_data: Dict со структурой SyncPolicyData для item'ов коллекций 68 override_for_collections: Dict для переопределения значений для отдельных коллекций 69 вида [collection_id, SyncPolicyData] 70 with_update: Флаг для возможности обновления существующих записей расширением. Может быть переопределён 71 в override_for_collections 72 with_delete: Флаг для возможности удаления существующих записей расширением. Может быть переопределён 73 в override_for_collections 74 Returns: 75 list[SyncPolicyItem] 76 """ 77 78 if not isinstance(override_for_collections, dict): 79 override_for_collections = {} 80 81 items = [] 82 83 for collection_id, collection_name in collections_map.items(): 84 if not collection_id: 85 continue 86 87 kwargs = { 88 "with_update": with_update, 89 "with_delete": with_delete, 90 "name": collection_name, 91 **override_for_collections.get(collection_id, sync_policies_data), 92 "collection_id": collection_id, 93 } 94 95 items.append(SyncPolicyItem(**kwargs)) 96 97 return items 98 99 100def get_extension_descriptor( 101 ext_host: str, ext_id: str, ext_name: str, ext_description: str, 102 ext_version: str, ext_version_description: str, 103 ext_deps: Optional[list[str]] = None 104) -> manager_service_pb2.ExtensionDescriptor: 105 """ 106 Функция для получения дескриптора расширения на основании его данных 107 108 Arguments: 109 ext_host (str): адрес расширения 110 ext_id (str): идентификатор расширения 111 ext_name (str): название расширения 112 ext_version (str): версия расширения 113 ext_version_description (str): описание расширения 114 ext_deps (list[str] | None): зависимости расширения 115 116 Returns: 117 manager_service_pb2.ExtensionDescriptor 118 """ 119 120 if not ext_deps: 121 ext_deps = [] 122 123 return manager_service_pb2.ExtensionDescriptor( 124 extension=ext_id, 125 title=ext_name, 126 description=ext_description, 127 version=ext_version, 128 version_description=ext_version_description, 129 deps=ext_deps, 130 url=ext_host 131 )
def
datasource_items_from_collections( collections_map: dict[str, str], datasource_data: perxis.extensions.item_models.DataSourceData, override_for_collections: dict[str, perxis.extensions.item_models.DataSourceData] | None = None, with_update: bool = True, with_delete: bool = True) -> list[perxis.extensions.item_models.DataSourceItem]:
12def datasource_items_from_collections( 13 collections_map: dict[str, str], 14 datasource_data: DataSourceData, 15 override_for_collections: dict[str, DataSourceData] | None = None, 16 with_update: bool = True, 17 with_delete: bool = True, 18) -> list[DataSourceItem]: 19 """ 20 Создание записей источников данных на базе маппинга коллекций. 21 22 Arguments: 23 collections_map: Dict вида [collection_id, collection_name] 24 datasource_data: Dict со структурой DataSourceData для item'ов коллекций 25 override_for_collections: Dict для переопределения значений для отдельных коллекций 26 вида [collection_id, DataSourceData] 27 with_update: Флаг для возможности обновления существующих записей расширением. Может быть переопределён 28 в override_for_collections 29 with_delete: Флаг для возможности удаления существующих записей расширением. Может быть переопределён 30 в override_for_collections 31 Returns: 32 list[DataSourceItem] 33 """ 34 35 if not isinstance(override_for_collections, dict): 36 override_for_collections = {} 37 38 items = [] 39 40 for collection_id in collections_map: 41 if not collection_id: 42 continue 43 44 kwargs = { 45 "with_update": with_update, 46 "with_delete": with_delete, 47 **override_for_collections.get(collection_id, datasource_data), 48 "collection_id": collection_id, 49 } 50 51 items.append(DataSourceItem(**kwargs)) 52 53 return items
Создание записей источников данных на базе маппинга коллекций.
Arguments:
- collections_map: Dict вида [collection_id, collection_name]
- datasource_data: Dict со структурой DataSourceData для item'ов коллекций
- override_for_collections: Dict для переопределения значений для отдельных коллекций вида [collection_id, DataSourceData]
- with_update: Флаг для возможности обновления существующих записей расширением. Может быть переопределён в override_for_collections
- with_delete: Флаг для возможности удаления существующих записей расширением. Может быть переопределён в override_for_collections
Returns:
list[DataSourceItem]
def
sync_policies_from_collections( collections_map: dict[str, str], sync_policies_data: perxis.extensions.item_models.SyncPolicyData, override_for_collections: dict[str, perxis.extensions.item_models.SyncPolicyData] | None = None, with_update: bool = True, with_delete: bool = True) -> list[perxis.extensions.item_models.SyncPolicyItem]:
56def sync_policies_from_collections( 57 collections_map: dict[str, str], 58 sync_policies_data: SyncPolicyData, 59 override_for_collections: dict[str, SyncPolicyData] | None = None, 60 with_update: bool = True, 61 with_delete: bool = True, 62) -> list[SyncPolicyItem]: 63 """ 64 Создание записей синхронизации коллекций на базе маппинга коллекций 65 66 Arguments: 67 collections_map: Dict вида [collection_id, collection_name] 68 sync_policies_data: Dict со структурой SyncPolicyData для item'ов коллекций 69 override_for_collections: Dict для переопределения значений для отдельных коллекций 70 вида [collection_id, SyncPolicyData] 71 with_update: Флаг для возможности обновления существующих записей расширением. Может быть переопределён 72 в override_for_collections 73 with_delete: Флаг для возможности удаления существующих записей расширением. Может быть переопределён 74 в override_for_collections 75 Returns: 76 list[SyncPolicyItem] 77 """ 78 79 if not isinstance(override_for_collections, dict): 80 override_for_collections = {} 81 82 items = [] 83 84 for collection_id, collection_name in collections_map.items(): 85 if not collection_id: 86 continue 87 88 kwargs = { 89 "with_update": with_update, 90 "with_delete": with_delete, 91 "name": collection_name, 92 **override_for_collections.get(collection_id, sync_policies_data), 93 "collection_id": collection_id, 94 } 95 96 items.append(SyncPolicyItem(**kwargs)) 97 98 return items
Создание записей синхронизации коллекций на базе маппинга коллекций
Arguments:
- collections_map: Dict вида [collection_id, collection_name]
- sync_policies_data: Dict со структурой SyncPolicyData для item'ов коллекций
- override_for_collections: Dict для переопределения значений для отдельных коллекций вида [collection_id, SyncPolicyData]
- with_update: Флаг для возможности обновления существующих записей расширением. Может быть переопределён в override_for_collections
- with_delete: Флаг для возможности удаления существующих записей расширением. Может быть переопределён в override_for_collections
Returns:
list[SyncPolicyItem]
def
get_extension_descriptor( ext_host: str, ext_id: str, ext_name: str, ext_description: str, ext_version: str, ext_version_description: str, ext_deps: Optional[list[str]] = None) -> extensions.manager_service_pb2.ExtensionDescriptor:
101def get_extension_descriptor( 102 ext_host: str, ext_id: str, ext_name: str, ext_description: str, 103 ext_version: str, ext_version_description: str, 104 ext_deps: Optional[list[str]] = None 105) -> manager_service_pb2.ExtensionDescriptor: 106 """ 107 Функция для получения дескриптора расширения на основании его данных 108 109 Arguments: 110 ext_host (str): адрес расширения 111 ext_id (str): идентификатор расширения 112 ext_name (str): название расширения 113 ext_version (str): версия расширения 114 ext_version_description (str): описание расширения 115 ext_deps (list[str] | None): зависимости расширения 116 117 Returns: 118 manager_service_pb2.ExtensionDescriptor 119 """ 120 121 if not ext_deps: 122 ext_deps = [] 123 124 return manager_service_pb2.ExtensionDescriptor( 125 extension=ext_id, 126 title=ext_name, 127 description=ext_description, 128 version=ext_version, 129 version_description=ext_version_description, 130 deps=ext_deps, 131 url=ext_host 132 )
Функция для получения дескриптора расширения на основании его данных
Arguments:
- ext_host (str): адрес расширения
- ext_id (str): идентификатор расширения
- ext_name (str): название расширения
- ext_version (str): версия расширения
- ext_version_description (str): описание расширения
- ext_deps (list[str] | None): зависимости расширения
Returns:
manager_service_pb2.ExtensionDescriptor