Skip to content
Snippets Groups Projects
Commit 93beed6e authored by Eterevskiy Georgiy's avatar Eterevskiy Georgiy
Browse files

Merge branch 'feature/AUTO-3946' into 'master'

Добавлена возможность указания data для SyncPolicyItem и DataSourceItem при создании с помощью функций

See merge request perxis/perxis-python!90
parents 595116f0 8d6bfc05
No related branches found
No related tags found
No related merge requests found
TEST_COLLECTION_ID = "test_collection"
ANOTHER_COLLECTION_ID = "another_test_collection"
schemes_mapping = {
TEST_COLLECTION_ID: "Тестовая коллекция",
ANOTHER_COLLECTION_ID: "Ещё одна тестовая коллекция",
}
from perxis.extensions.item_models import DataSourceData, SyncPolicyData
DATASOURCE_DATA: DataSourceData = dict(
query="Some query",
content_type="json",
)
DATASOURCE_DATA_FOR_ANOTHER_COLLECTION: DataSourceData = dict(
query="Overrided some query",
with_update=False,
with_delete=False,
)
SYNC_POLICY_DATA: SyncPolicyData = dict(
key="id",
export_view=True,
)
SYNC_POLICY_DATA_ANOTHER_COLLECTION: SyncPolicyData = dict(
key="overrided",
export_view=False,
with_update=True,
with_delete=True,
)
\ No newline at end of file
{
"ui": {
"options": {
"fields": [
"aaa",
"bbb"
]
}
},
"type": "object",
"params": {
"inline": false,
"fields": {
"bbb": {
"title": "Ещё поле",
"ui": {
"widget": "StringInput"
},
"type": "string",
"params": {}
},
"aaa": {
"title": "Какое то поле",
"ui": {
"widget": "StringInput"
},
"type": "string",
"params": {}
}
}
},
"loaded": false
}
......@@ -2,6 +2,8 @@ import random
from constants import extension
from constants import collections as extension_collections
from items import (SYNC_POLICY_DATA, SYNC_POLICY_DATA_ANOTHER_COLLECTION,
DATASOURCE_DATA, DATASOURCE_DATA_FOR_ANOTHER_COLLECTION)
from perxis.extensions.utils import datasource_items_from_collections, sync_policies_from_collections
......@@ -78,9 +80,27 @@ class Servicer(ExtensionService):
)
]
# items = (datasource_items_from_collections(extension_collections.schemes_mapping)
# + sync_policies_from_collections(extension_collections.schemes_mapping))
items = (
datasource_items_from_collections(
collections_map=extension_collections.schemes_mapping,
datasource_data=DATASOURCE_DATA,
override_for_collections={
extension_collections.ANOTHER_COLLECTION_ID: DATASOURCE_DATA_FOR_ANOTHER_COLLECTION
},
with_update=True,
with_delete=True
)
+ sync_policies_from_collections(
collections_map=extension_collections.schemes_mapping,
sync_policies_data=SYNC_POLICY_DATA,
override_for_collections={
extension_collections.ANOTHER_COLLECTION_ID: SYNC_POLICY_DATA_ANOTHER_COLLECTION,
},
with_delete=False,
with_update=False,
)
)
'''
items = [
DataSourceItem(
collection_id="test_collection"
......@@ -113,6 +133,7 @@ class Servicer(ExtensionService):
]
)
]
'''
async def action_get_current_organization_and_users(
self,
......
import abc
from typing import TypedDict, NotRequired
from google.protobuf.struct_pb2 import Struct
from perxis.extensions.item_rules import AbstractRule, IfCollectionExists, IfExtensionInstalled
class DataSourceData(TypedDict):
query: NotRequired[str]
content_type: NotRequired[str]
exclude: NotRequired[bool]
with_update: NotRequired[bool]
with_delete: NotRequired[bool]
class SyncPolicyData(TypedDict):
key: NotRequired[str]
export_view: NotRequired[bool]
remove_collection: NotRequired[bool]
deny_publish: NotRequired[bool]
deny_delete: NotRequired[bool]
deny_create: NotRequired[bool]
deny_read: NotRequired[bool]
hidden: NotRequired[bool]
with_update: NotRequired[bool]
with_delete: NotRequired[bool]
class AbstractItem(metaclass=abc.ABCMeta):
"""
Абстрактный класс для item'а. Нужен для определения общих свойств без реализации какого
......@@ -109,7 +131,6 @@ class SyncPolicyItem(AbstractItem):
"""
Класс для коллекции hoop_item_sync_policies расширения perxishoop
"""
identifier_field = "collection"
def __init__(
......@@ -117,16 +138,16 @@ class SyncPolicyItem(AbstractItem):
collection_id: str,
name: str = "",
key: str = "id",
export_view: bool = True,
export_view: bool = False,
remove_collection: bool = False,
deny_publish: bool = True,
deny_delete: bool = True,
deny_create: bool = True,
deny_publish: bool = False,
deny_delete: bool = False,
deny_create: bool = False,
deny_read: bool = False,
hidden: bool = False,
rules: list[AbstractRule] | None = None,
with_update: bool = True,
with_delete: bool = True,
with_update: bool = False,
with_delete: bool = False,
):
self.collection_id = "hoop_item_sync_policies"
self.rules = rules or [IfExtensionInstalled("perxishoop")]
......
from typing import Optional
from perxis.extensions import manager_service_pb2
from perxis.extensions.item_models import DataSourceItem, SyncPolicyItem
from perxis.extensions.item_models import DataSourceItem, DataSourceData, SyncPolicyItem, SyncPolicyData
def datasource_items_from_collections(
collections_map: dict[str, str],
datasource_data: DataSourceData,
override_for_collections: dict[str, DataSourceData] | None = None,
with_update: bool = True,
with_delete: bool = True,
) -> list[DataSourceItem]:
"""
Создание записей источников данных на базе маппинга коллекций
Создание записей источников данных на базе маппинга коллекций.
Args:
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]
"""
return [
DataSourceItem(
collection_id=collection_id,
with_delete=with_delete,
with_update=with_update
)
for collection_id in collections_map
if collection_id
]
if not isinstance(override_for_collections, dict):
override_for_collections = {}
items = []
for collection_id in collections_map:
if not collection_id:
continue
kwargs = {
"with_update": with_update,
"with_delete": with_delete,
**override_for_collections.get(collection_id, datasource_data),
"collection_id": collection_id,
}
items.append(DataSourceItem(**kwargs))
return items
def sync_policies_from_collections(
collections_map: dict[str, str],
sync_policies_data: SyncPolicyData,
override_for_collections: dict[str, SyncPolicyData] | None = None,
with_update: bool = True,
with_delete: bool = True,
) -> list[SyncPolicyItem]:
"""
Создание записей синхронизации коллекций на базе маппинга коллекций
Args:
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]
"""
return [
SyncPolicyItem(
collection_id=collection_id,
name=collection_name,
with_update=with_update,
with_delete=with_delete,
)
for collection_id, collection_name in collections_map.items()
if collection_id
]
if not isinstance(override_for_collections, dict):
override_for_collections = {}
items = []
for collection_id, collection_name in collections_map.items():
if not collection_id:
continue
kwargs = {
"with_update": with_update,
"with_delete": with_delete,
"name": collection_name,
**override_for_collections.get(collection_id, sync_policies_data),
"collection_id": collection_id,
}
items.append(SyncPolicyItem(**kwargs))
return items
def get_extension_descriptor(
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment