perxis.extensions.item_rules

  1import abc
  2import typing
  3import logging
  4
  5from perxis.collections import collections_pb2
  6from perxis.extensions import manager_service_pb2
  7
  8
  9if typing.TYPE_CHECKING:
 10    from perxis.extensions.item_models import AbstractItem
 11
 12
 13class AbstractRule(metaclass=abc.ABCMeta):
 14    """
 15        Абстрактное правило для обработки создания item'ов в системе. Правила могут
 16        использоваться в кейсах:
 17            1. В случае если item опционален и невозможность его создания не приведёт к невозможности
 18               использовать само расширение
 19            2. Если при создании item'а нужно обогатить его данные чем-то внешним, не относящимся к
 20               текущему расширению. Например - при создании item'ов для цепочки сервиса уведомлений.
 21               Каждый последующий item может с помощью правил получать результат создания предыдущего
 22               и на базе этих данных формировать ref
 23
 24        Так как сервисов в ExtensionService очень много, для того чтобы код их проброса в объект rule
 25        не был излишне длинным - это делается неявным пробросом свойств из ExtensionService. Таким
 26        образом и код короче получается, и в случае если будут добавлены какие то новые сервисы они
 27        автоматически будут проброшены сюда.
 28
 29        Attributes:
 30            channel (grpc.Channel): ссылка на grpc канал
 31            *_service: ссылки на сервисы. Именование аналогично тому как они указаны в классе ExtensionService
 32    """
 33
 34    @property
 35    def rule_name(self):
 36        """
 37            Название правила
 38
 39            Returns:
 40                  str
 41        """
 42        return self.__class__.__name__
 43
 44    @property
 45    def logger(self):
 46        """
 47            Логгер для правила
 48
 49            Returns:
 50                  logging.Logger
 51        """
 52        return logging.getLogger(self.rule_name)
 53
 54    def bind_services(self, service_list: list[tuple[str, object]]):
 55        """
 56            Нужно для автоматического подключения сервисов из ExtensionService. Проблема
 57            в том что их очень много и если руками явно указывать - будет большой boilerplate
 58
 59            Arguments:
 60                service_list (list[tuple[str, object]]): список доступных сервисов и их идентифиакторы
 61        """
 62
 63        for service_name, service in service_list:
 64            setattr(self, service_name, service)
 65
 66    @abc.abstractmethod
 67    async def act(self, item: "AbstractItem", space_id: str, env_id: str):
 68        """
 69            Абстрактный метод для указания логики правила. Обработка исключений ведётся в методе __call__,
 70            уровнем выше. Если метод `act` отработал без исключений считается что правило успешно применено
 71
 72            Arguments:
 73                 item (AbstractItem): item с которым будет вестись работа
 74                 space_id (str): идентификатор пространства
 75                 env_id (str): идентификатор окружения
 76        """
 77
 78        raise NotImplementedError()
 79
 80    async def __call__(self, item: "AbstractItem", space_id: str, env_id: str) -> bool:
 81        try:
 82            await self.act(item, space_id, env_id)
 83
 84            return True
 85        except Exception as e:
 86            self.logger.warning(
 87                f"Правило {self.rule_name} не "
 88                f"сработало для {item.collection_id} (item {item.identifier}), "
 89                f"ошибка - {e}"
 90            )
 91
 92            return False
 93
 94
 95class IfCollectionExists(AbstractRule):
 96    """
 97        Правило для опциональных item'ов которые нужно создавать в случае наличия
 98        определённой коллекции в perxis. Может быть полезно если коллекция широко используется
 99        но в данный момент устанавливается только вручную и не привязана ни к одному из расширений
100    """
101
102    async def act(self, item: "AbstractItem", space_id: str, env_id: str):
103        """
104            Проверка наличия коллекции в perxis для указанного пространства и окружения
105
106            Arguments:
107                 item (AbstractItem): item с которым будет вестись работа
108                 space_id (str): идентификатор пространства
109                 env_id (str): идентификатор окружения
110        """
111
112        message = await self.collections_service.Get(
113            collections_pb2.GetRequest(
114                space_id=space_id,
115                env_id=env_id,
116                collection_id=item.collection_id
117            )
118        )
119
120        if not message.collection:
121            raise ValueError(f"Коллекция {item.collection_id} не найдена")
122
123
124class IfExtensionInstalled(AbstractRule):
125    """
126        Правило для опциональных item'ов которые нужно создавать только в случае
127        наличия определённого расширения в perxis
128    """
129
130    required_extension: str
131
132    def __init__(self, required_extension: str):
133        """
134            Arguments:
135                required_extension (str): Идентификатор требуемоого расширения
136        """
137        self.required_extension = required_extension
138
139    async def act(self, item: "AbstractItem", space_id: str, env_id: str):
140        """
141            Проверка наличия установленного расширения в perxis для указанного пространства и окружения
142
143            Arguments:
144                 item (AbstractItem): item с которым будет вестись работа
145                 space_id (str): идентификатор пространства
146                 env_id (str): идентификатор окружения
147        """
148
149        registered_extensions = await self.ext_manager_service.ListRegisteredExtensions(
150            manager_service_pb2.ListRegisteredExtensionsRequest()
151        )
152
153        ext_is_installed = False
154        for ext in registered_extensions.extensions:
155            if ext.extension == self.required_extension:
156                ext_is_installed = True
157                break
158
159        if not ext_is_installed:
160            raise ValueError(
161                f"Расширение {self.required_extension} не установлено в perxis"
162            )
class AbstractRule:
14class AbstractRule(metaclass=abc.ABCMeta):
15    """
16        Абстрактное правило для обработки создания item'ов в системе. Правила могут
17        использоваться в кейсах:
18            1. В случае если item опционален и невозможность его создания не приведёт к невозможности
19               использовать само расширение
20            2. Если при создании item'а нужно обогатить его данные чем-то внешним, не относящимся к
21               текущему расширению. Например - при создании item'ов для цепочки сервиса уведомлений.
22               Каждый последующий item может с помощью правил получать результат создания предыдущего
23               и на базе этих данных формировать ref
24
25        Так как сервисов в ExtensionService очень много, для того чтобы код их проброса в объект rule
26        не был излишне длинным - это делается неявным пробросом свойств из ExtensionService. Таким
27        образом и код короче получается, и в случае если будут добавлены какие то новые сервисы они
28        автоматически будут проброшены сюда.
29
30        Attributes:
31            channel (grpc.Channel): ссылка на grpc канал
32            *_service: ссылки на сервисы. Именование аналогично тому как они указаны в классе ExtensionService
33    """
34
35    @property
36    def rule_name(self):
37        """
38            Название правила
39
40            Returns:
41                  str
42        """
43        return self.__class__.__name__
44
45    @property
46    def logger(self):
47        """
48            Логгер для правила
49
50            Returns:
51                  logging.Logger
52        """
53        return logging.getLogger(self.rule_name)
54
55    def bind_services(self, service_list: list[tuple[str, object]]):
56        """
57            Нужно для автоматического подключения сервисов из ExtensionService. Проблема
58            в том что их очень много и если руками явно указывать - будет большой boilerplate
59
60            Arguments:
61                service_list (list[tuple[str, object]]): список доступных сервисов и их идентифиакторы
62        """
63
64        for service_name, service in service_list:
65            setattr(self, service_name, service)
66
67    @abc.abstractmethod
68    async def act(self, item: "AbstractItem", space_id: str, env_id: str):
69        """
70            Абстрактный метод для указания логики правила. Обработка исключений ведётся в методе __call__,
71            уровнем выше. Если метод `act` отработал без исключений считается что правило успешно применено
72
73            Arguments:
74                 item (AbstractItem): item с которым будет вестись работа
75                 space_id (str): идентификатор пространства
76                 env_id (str): идентификатор окружения
77        """
78
79        raise NotImplementedError()
80
81    async def __call__(self, item: "AbstractItem", space_id: str, env_id: str) -> bool:
82        try:
83            await self.act(item, space_id, env_id)
84
85            return True
86        except Exception as e:
87            self.logger.warning(
88                f"Правило {self.rule_name} не "
89                f"сработало для {item.collection_id} (item {item.identifier}), "
90                f"ошибка - {e}"
91            )
92
93            return False

Абстрактное правило для обработки создания item'ов в системе. Правила могут использоваться в кейсах: 1. В случае если item опционален и невозможность его создания не приведёт к невозможности использовать само расширение 2. Если при создании item'а нужно обогатить его данные чем-то внешним, не относящимся к текущему расширению. Например - при создании item'ов для цепочки сервиса уведомлений. Каждый последующий item может с помощью правил получать результат создания предыдущего и на базе этих данных формировать ref

Так как сервисов в ExtensionService очень много, для того чтобы код их проброса в объект rule не был излишне длинным - это делается неявным пробросом свойств из ExtensionService. Таким образом и код короче получается, и в случае если будут добавлены какие то новые сервисы они автоматически будут проброшены сюда.

Attributes:
  • channel (grpc.Channel): ссылка на grpc канал
  • *_service: ссылки на сервисы. Именование аналогично тому как они указаны в классе ExtensionService
rule_name
35    @property
36    def rule_name(self):
37        """
38            Название правила
39
40            Returns:
41                  str
42        """
43        return self.__class__.__name__

Название правила

Returns:

str

logger
45    @property
46    def logger(self):
47        """
48            Логгер для правила
49
50            Returns:
51                  logging.Logger
52        """
53        return logging.getLogger(self.rule_name)

Логгер для правила

Returns:

logging.Logger

def bind_services(self, service_list: list[tuple[str, object]]):
55    def bind_services(self, service_list: list[tuple[str, object]]):
56        """
57            Нужно для автоматического подключения сервисов из ExtensionService. Проблема
58            в том что их очень много и если руками явно указывать - будет большой boilerplate
59
60            Arguments:
61                service_list (list[tuple[str, object]]): список доступных сервисов и их идентифиакторы
62        """
63
64        for service_name, service in service_list:
65            setattr(self, service_name, service)

Нужно для автоматического подключения сервисов из ExtensionService. Проблема в том что их очень много и если руками явно указывать - будет большой boilerplate

Arguments:
  • service_list (list[tuple[str, object]]): список доступных сервисов и их идентифиакторы
@abc.abstractmethod
async def act( self, item: perxis.extensions.item_models.AbstractItem, space_id: str, env_id: str):
67    @abc.abstractmethod
68    async def act(self, item: "AbstractItem", space_id: str, env_id: str):
69        """
70            Абстрактный метод для указания логики правила. Обработка исключений ведётся в методе __call__,
71            уровнем выше. Если метод `act` отработал без исключений считается что правило успешно применено
72
73            Arguments:
74                 item (AbstractItem): item с которым будет вестись работа
75                 space_id (str): идентификатор пространства
76                 env_id (str): идентификатор окружения
77        """
78
79        raise NotImplementedError()

Абстрактный метод для указания логики правила. Обработка исключений ведётся в методе __call__, уровнем выше. Если метод act отработал без исключений считается что правило успешно применено

Arguments:
  • item (AbstractItem): item с которым будет вестись работа
  • space_id (str): идентификатор пространства
  • env_id (str): идентификатор окружения
class IfCollectionExists(AbstractRule):
 96class IfCollectionExists(AbstractRule):
 97    """
 98        Правило для опциональных item'ов которые нужно создавать в случае наличия
 99        определённой коллекции в perxis. Может быть полезно если коллекция широко используется
100        но в данный момент устанавливается только вручную и не привязана ни к одному из расширений
101    """
102
103    async def act(self, item: "AbstractItem", space_id: str, env_id: str):
104        """
105            Проверка наличия коллекции в perxis для указанного пространства и окружения
106
107            Arguments:
108                 item (AbstractItem): item с которым будет вестись работа
109                 space_id (str): идентификатор пространства
110                 env_id (str): идентификатор окружения
111        """
112
113        message = await self.collections_service.Get(
114            collections_pb2.GetRequest(
115                space_id=space_id,
116                env_id=env_id,
117                collection_id=item.collection_id
118            )
119        )
120
121        if not message.collection:
122            raise ValueError(f"Коллекция {item.collection_id} не найдена")

Правило для опциональных item'ов которые нужно создавать в случае наличия определённой коллекции в perxis. Может быть полезно если коллекция широко используется но в данный момент устанавливается только вручную и не привязана ни к одному из расширений

async def act( self, item: perxis.extensions.item_models.AbstractItem, space_id: str, env_id: str):
103    async def act(self, item: "AbstractItem", space_id: str, env_id: str):
104        """
105            Проверка наличия коллекции в perxis для указанного пространства и окружения
106
107            Arguments:
108                 item (AbstractItem): item с которым будет вестись работа
109                 space_id (str): идентификатор пространства
110                 env_id (str): идентификатор окружения
111        """
112
113        message = await self.collections_service.Get(
114            collections_pb2.GetRequest(
115                space_id=space_id,
116                env_id=env_id,
117                collection_id=item.collection_id
118            )
119        )
120
121        if not message.collection:
122            raise ValueError(f"Коллекция {item.collection_id} не найдена")

Проверка наличия коллекции в perxis для указанного пространства и окружения

Arguments:
  • item (AbstractItem): item с которым будет вестись работа
  • space_id (str): идентификатор пространства
  • env_id (str): идентификатор окружения
class IfExtensionInstalled(AbstractRule):
125class IfExtensionInstalled(AbstractRule):
126    """
127        Правило для опциональных item'ов которые нужно создавать только в случае
128        наличия определённого расширения в perxis
129    """
130
131    required_extension: str
132
133    def __init__(self, required_extension: str):
134        """
135            Arguments:
136                required_extension (str): Идентификатор требуемоого расширения
137        """
138        self.required_extension = required_extension
139
140    async def act(self, item: "AbstractItem", space_id: str, env_id: str):
141        """
142            Проверка наличия установленного расширения в perxis для указанного пространства и окружения
143
144            Arguments:
145                 item (AbstractItem): item с которым будет вестись работа
146                 space_id (str): идентификатор пространства
147                 env_id (str): идентификатор окружения
148        """
149
150        registered_extensions = await self.ext_manager_service.ListRegisteredExtensions(
151            manager_service_pb2.ListRegisteredExtensionsRequest()
152        )
153
154        ext_is_installed = False
155        for ext in registered_extensions.extensions:
156            if ext.extension == self.required_extension:
157                ext_is_installed = True
158                break
159
160        if not ext_is_installed:
161            raise ValueError(
162                f"Расширение {self.required_extension} не установлено в perxis"
163            )

Правило для опциональных item'ов которые нужно создавать только в случае наличия определённого расширения в perxis

IfExtensionInstalled(required_extension: str)
133    def __init__(self, required_extension: str):
134        """
135            Arguments:
136                required_extension (str): Идентификатор требуемоого расширения
137        """
138        self.required_extension = required_extension
Arguments:
  • required_extension (str): Идентификатор требуемоого расширения
required_extension: str
async def act( self, item: perxis.extensions.item_models.AbstractItem, space_id: str, env_id: str):
140    async def act(self, item: "AbstractItem", space_id: str, env_id: str):
141        """
142            Проверка наличия установленного расширения в perxis для указанного пространства и окружения
143
144            Arguments:
145                 item (AbstractItem): item с которым будет вестись работа
146                 space_id (str): идентификатор пространства
147                 env_id (str): идентификатор окружения
148        """
149
150        registered_extensions = await self.ext_manager_service.ListRegisteredExtensions(
151            manager_service_pb2.ListRegisteredExtensionsRequest()
152        )
153
154        ext_is_installed = False
155        for ext in registered_extensions.extensions:
156            if ext.extension == self.required_extension:
157                ext_is_installed = True
158                break
159
160        if not ext_is_installed:
161            raise ValueError(
162                f"Расширение {self.required_extension} не установлено в perxis"
163            )

Проверка наличия установленного расширения в perxis для указанного пространства и окружения

Arguments:
  • item (AbstractItem): item с которым будет вестись работа
  • space_id (str): идентификатор пространства
  • env_id (str): идентификатор окружения