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