Исходный код EJIO.utils.interface.disposable

## /utils/interface/disposable.py

import typing
from typing import Any, Protocol, runtime_checkable
from loguru import logger

__all__ = [
    'Disposable',
    'DisposableMeta',
]




class DisposableMeta(typing._ProtocolMeta):
    """
    Метакласс для автоматического рантайм-учета всех уничтожаемых
    и очищаемых структур.
    """
    __slots__ = ()

    _registry: list[type] = []

    def __new__(
            mcs,
            name: str,
            bases: tuple[type, ...],
            attrs: dict[str, Any]
    ) -> type:
        """
        Добавляет на учет все классы, метаклассом которых является DisposableMeta.

        :param name: Имя создаваемого класса.
        :param bases: Список классов родителей.
        :param attrs: Словарь методов и аттрибутов класса.
        :return: Новый класс - наследник DisposableMeta.
        """
        cls = super().__new__(mcs, name, bases, attrs)

        # Пропускаем сам базовый протокол Disposable
        if name != 'Disposable':
            # Проверяем наличие методов clear или destroy в атрибутах или у предков
            has_clear = 'clear' in attrs or any(hasattr(b, 'clear') for b in bases)
            has_destroy = 'destroy' in attrs or any(hasattr(b, 'destroy') for b in bases)

            # Проверяем, реализует ли класс метод clear или destroy
            if has_clear or has_destroy:
                if cls not in mcs._registry:
                    mcs._registry.append(cls)
                    logger.debug(f"Struct '{name}' taken into automatic runtime accounting Disposable.")
        return cls



    @classmethod
    def dispose_all(mcs) -> None:
        """
        Централизованно и каскадно уничтожает и очищает абсолютно все
        зарегистрированные структуры.
        """
        logger.info(
            f"Automatic global runtime cleaning launched of ({len(mcs._registry)} Disposable structures)...")

        for cls_target in list(mcs._registry):
            cleaned = False

            try:
                # 1. Если у структуры объявлен метод clear
                if hasattr(cls_target, 'clear') and callable(getattr(cls_target, 'clear')):
                    cls_target.clear()
                    logger.debug(f"Auto-cleaning: structure'{cls_target.__name__}' method .clear() successfully completed.")
                    cleaned = True

                # 2. Если у структуры объявлен метод destroy
                elif hasattr(cls_target, 'destroy') and callable(getattr(cls_target, 'destroy')):
                    cls_target.destroy()
                    logger.debug(f"Auto-cleaning: structure '{cls_target.__name__}' method .destroy() successfully completed.")
                    cleaned = True

                if not cleaned:
                    raise AttributeError(f"Struct '{cls_target.__name__}' does not have a callable attribute 'clear' or 'destroy'.")

            except Exception as e:
                logger.error(f"Error auto-clearing structure '{cls_target.__name__}': {e}")

        logger.info("Global auto-clear runtime Disposable completed.")






@runtime_checkable
class Disposable(Protocol, metaclass=DisposableMeta):
    """
    Единый интерфейс-протокол для всех долгоживущих рантайм-структур.
    Гарантирует автоматическую очистку памяти при завершении работы приложения.
    """
    __slots__ = ()



    def clear(self) -> None:
        """
        Метод полной очистки внутреннего хранилища/состояния.

        :raises NotImplementedError: должен быть реализован в подклассах.
        """
        raise NotImplementedError




    def destroy(self) -> None:
        """
        Метод полной очистки внутреннего хранилища/состояния.

        :raises NotImplementedError: должен быть реализован в подклассах.
        """
        raise NotImplementedError