## /utils/struct/registry.py
from typing import Any, Callable, get_args
from loguru import logger
from ..interface import StaticClass, IsolatedStorageMixin, Disposable
from .err import RegistryKeyError, RegistryTypeError
__all__ = [
'Registry',
'create_decorator'
]
[документация]
class Registry[T](IsolatedStorageMixin[T], StaticClass, Disposable):
"""
Абстрактный базовый интерфейс для конкретных классов-регистров.
Объединяет запрет на создание экземпляров и механизм изолированного хранилища.
Предоставляет API для регистрации и получения классов по строковому ключу.
Работает строго с типами, унаследованными от T или являющимися T.
"""
__slots__ = ()
[документация]
@classmethod
def register[S](cls, key: str, target_cls: type[S]) -> type[S]:
"""
Регистрирует класс или объект в текущем регистре под определенным ключом.
Выполняет потокобезопасную проверку на уникальность ключа.
:param key: Уникальный строковый идентификатор для класса.
:param target_cls: Класс, который необходимо зарегистрировать.
:return: Сам зарегистрированный класс (поддерживает синтаксис декоратора).
:raises RegistryKeyError: Если ключ уже занят в этом конкретном регистре.
"""
# Динамическое извлечение реального типа T в рантайме
expected_type: type[Any] | None = None
# Перебираем базовые классы конкретного регистра (наследника)
# Ищем, от какого именно Registry[КонкретныйКласс] он унаследован
for base in getattr(cls, '__orig_bases__', []):
if hasattr(base, '__origin__') and issubclass(base.__origin__, Registry):
args = get_args(base)
if args:
expected_type = args[0]
break
# Прямая рантайм-проверка иерархии классов
# Проверяем, только если регистр был объявлен с конкретным типом, отличным от object
if isinstance(target_cls, type):
if expected_type and expected_type is not object and expected_type is not Any:
try:
# Попытка стандартной проверки (работает для ABC, обычных классов и @runtime_checkable протоколов)
is_child = issubclass(target_cls, expected_type)
except TypeError:
# Сюда мы попадаем, если expected_type — это Protocol БЕЗ декоратора @runtime_checkable.
# Реализуем строгую валидацию по структуре: проверяем, что target_cls
# содержит все абстрактные и публичные методы, объявленные в протоколе.
is_child = True
# Извлекаем все ожидаемые методы и атрибуты протокола (исключая системные вроде __abstractmethods__)
expected_methods = {
attr for attr in dir(expected_type)
if not attr.startswith('__') or attr in expected_type.__abstractmethods__ #type: ignore
}
for method in expected_methods:
if not hasattr(target_cls, method):
is_child = False
break
if not is_child:
raise RegistryTypeError(
f"Error in '{cls.__name__}': Class '{target_cls.__name__}' "
f"is not a child of a required interface '{expected_type.__name__}'."
)
with cls._LOCK:
if key in cls._STORAGE: # type: ignore
existing = cls._STORAGE[key] # type: ignore
# Формируем красивое имя для логирования ошибок в зависимости от того, класс это или объект
exist_name = existing.__name__ if isinstance(existing, type) else type(existing).__name__
new_name = target_cls.__name__ if isinstance(target_cls, type) else type(target_cls).__name__
raise RegistryKeyError(
f"Conflict in '{cls.__name__}': Key '{key}' already registered "
f"for element '{exist_name}'. "
f"Attempt to rewrite it by element '{new_name}' declined."
)
cls._STORAGE[key] = target_cls # type: ignore
logger.debug(f"Key {key} successfully registered in register '{cls.__name__}'.")
return target_cls
[документация]
@classmethod
def get(cls, key: str) -> type[T]:
"""
Возвращает зарегистрированный класс по его ключу из текущего регистра.
:param key: Строковый ключ.
:return: Класс, сохраненный под этим ключом.
:raises RegistryKeyError: Если класс с таким ключом не найден.
"""
with cls._LOCK:
if key not in cls._STORAGE: # type: ignore
raise RegistryKeyError(
f"In register '{cls.__name__}' class with key '{key}' not found."
)
return cls._STORAGE[key] # type: ignore
[документация]
@classmethod
def get_all(cls) -> dict[str, type[T]]:
"""
Возвращает изолированную копию словаря со всеми зарегистрированными классами.
:return: Словарь вида {ключ: класс}.
"""
with cls._LOCK:
return cls._STORAGE.copy() # type: ignore
[документация]
@classmethod
def has_key(cls, key: str) -> bool:
"""
Потокобезопасно проверяет, зарегистрирован ли какой-либо класс под данным ключом.
:param key: Строковый ключ для проверки.
:return: True, если ключ существует в реестре, иначе False.
"""
with cls._LOCK: # type: ignore[attr-defined]
return key in cls._STORAGE # type: ignore[attr-defined]
[документация]
@classmethod
def unregister(cls, key: str) -> None:
"""
Потокобезопасно удаляет зарегистрированный элемент из реестра по его ключу.
:param key: Строковый ключ для удаления.
:raises RegistryKeyError: Если ключ отсутствует в реестре.
"""
with cls._LOCK: # type: ignore[attr-defined]
if key not in cls._STORAGE: # type: ignore[attr-defined]
raise RegistryKeyError( # type: ignore[name-defined]
f"In register '{cls.__name__}' key '{key}' not found for unregistration."
)
del cls._STORAGE[key] # type: ignore[attr-defined]
logger.debug(f"Key {key} successfully unregistered in register '{cls.__name__}'.")
[документация]
@classmethod
def clear(cls) -> None:
"""
Потокобезопасно и полностью очищает внутреннее изолированное хранилище реестра.
Выгружает все удерживаемые ссылки из памяти.
"""
with cls._LOCK: # type: ignore[attr-defined]
cls._STORAGE.clear() # type: ignore[attr-defined]
logger.debug(f"Register '{cls.__name__}' completely cleared.")
[документация]
def create_decorator[R: Registry[Any]](registry_cls: type[R]) -> Callable[[str], Callable[[type[Any]], type[Any]]]:
"""
Фабрика, создающая именной декоратор для конкретного класса регистра.
Позволяет вместо длинного `@register_to(MyRegistry, "key")`
писать лаконичный `@register_my_registry("key")`.
:param registry_cls: Клас реестра для которого требуется создать декаратор.
:return: Кастомный декоратор регистрации объетов.
"""
def custom_registry_decorator(key: str) -> Callable[[type[Any]], type[Any]]:
"""
Кастомный декоратор для регистрации классов в реестр.
:param key: Ключ, под которым объект будет храниться в реестре.
:return: Обертку, выполняющую регистрацию класса.
"""
def decorator[E](target_cls: type[E]) -> type[E]:
"""
Обертка, выполняющая регистрацию класса в реестр.
:param target_cls: Класс, который нужно зарегестрировать.
:return: Класс, который нужно зарегестрировать.
"""
registry_cls.register(key, target_cls)
return target_cls
return decorator #type: ignore[return-value]
# Меняем метаданные самого декоратора
custom_registry_decorator.__name__ = f"register_to_{registry_cls.__name__.lower()}"
custom_registry_decorator.__doc__ = (
f"Специализированный декоратор для регистрации классов в реестре {registry_cls.__name__}.\n\n"
f":param key: Уникальный строковый ключ-идентификатор для регистрации.\n"
f":return: Возвращает исходный класс без изменений."
)
return custom_registry_decorator #type: ignore[return-value]