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 )
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
35 @property 36 def rule_name(self): 37 """ 38 Название правила 39 40 Returns: 41 str 42 """ 43 return self.__class__.__name__
Название правила
Returns:
str
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
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]]): список доступных сервисов и их идентифиакторы
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): идентификатор окружения
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. Может быть полезно если коллекция широко используется но в данный момент устанавливается только вручную и не привязана ни к одному из расширений
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): идентификатор окружения
Inherited Members
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
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): Идентификатор требуемоого расширения
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): идентификатор окружения