Skip to content
Snippets Groups Projects

Добавлена возможность установки item'ов

Merged Podosochnyy Maxim requested to merge feature/AUTO-3946 into master
+ 148
0
import abc
from google.protobuf.struct_pb2 import Struct
from perxis.extensions.item_rules import AbstractRule, IfCollectionExists, IfExtensionInstalled
class AbstractItem(metaclass=abc.ABCMeta):
"""
Абстрактный класс для item'а. Нужен для определения общих свойств без реализации какого
то конкретного конструктора.
"""
collection_id: str
data: dict
rules: list[AbstractRule]
identifier_field: str = "id"
with_update: bool
with_delete: bool
@property
def identifier(self):
return self.data[self.identifier_field]
@property
def struct(self) -> Struct:
s = Struct()
s.update(self.data)
return s
async def all_rules_is_satisfied(self, space_id: str, env_id: str) -> bool:
return all(
[
await rule(item=self, space_id=space_id, env_id=env_id)
for rule
in self.rules
]
)
def merge_data(self, data_from_perxis: dict) -> Struct:
"""
Мерж данных из перксиса и расширения. В приоритете данные расширения, если что-то
из них было вручную изменено в perxis - значение будет затёрто
"""
s = Struct()
s.update({
**data_from_perxis,
**self.data,
})
return s
class Item(AbstractItem):
"""
Общий класс для любого Item
"""
def __init__(
self,
collection_id: str,
data: dict,
with_update: bool = True,
with_delete: bool = True,
identifier_field: str = "id",
rules: list[AbstractRule] | None = None
):
self.collection_id = collection_id
self.data = data
self.identifier_field = identifier_field
self.rules = rules or []
self.with_update = with_update
self.with_delete = with_delete
class DataSourceItem(AbstractItem):
"""
Класс для системной коллекции web_datasources
"""
def __init__(
self,
collection_id: str,
rules: list[AbstractRule] | None = None,
query: str = "",
content_type: str = "",
exclude: bool = False,
with_update: bool = True,
with_delete: bool = True,
):
self.collection_id = "web_datasources"
self.rules = rules or [IfCollectionExists()]
self.data = {
"id": collection_id,
"collection": collection_id,
"query": query,
"content_type": content_type,
"exclude": exclude,
}
self.with_update = with_update
self.with_delete = with_delete
class SyncPolicyItem(AbstractItem):
"""
Класс для коллекции hoop_item_sync_policies расширения perxishoop
"""
identifier_field = "collection"
def __init__(
self,
collection_id: str,
name: str = "",
key: str = "id",
export_view: bool = True,
remove_collection: bool = False,
deny_publish: bool = True,
deny_delete: bool = True,
deny_create: bool = True,
deny_read: bool = False,
hidden: bool = False,
rules: list[AbstractRule] | None = None,
with_update: bool = True,
with_delete: bool = True,
):
self.collection_id = "hoop_item_sync_policies"
self.rules = rules or [IfExtensionInstalled("perxishoop")]
self.data = {
"name": name or f"Коллекция {collection_id}",
"collection": collection_id,
"key": key,
"export_view": export_view,
"remove_collection": remove_collection,
"deny_publish": deny_publish,
"deny_delete": deny_delete,
"deny_create": deny_create,
"deny_read": deny_read,
"hidden": hidden,
}
self.with_update = with_update
self.with_delete = with_delete
\ No newline at end of file
Loading