Source code for core_mixins.interfaces.factory

# -*- coding: utf-8 -*-

"""
Base interface for classes that acts like factories.
"""

import threading
from abc import ABC, abstractmethod
from inspect import isabstract
from typing import Type, Dict, Set, Optional

from core_mixins.compatibility import Self




class IFactory(ABC):
    """
    Base interface for classes that acts like
    factories. The IFactory pattern
    allows you to:

      - Automatically register concrete implementations.
      - Discover available implementations at runtime.
      - Create instances by string reference.
      - Maintain separate registries for different factory types.

    ``IFactory`` is a pure protocol, it provides the registration
    mechanism but does **not** store any references itself. Registries
    live exclusively on concrete sub-hierarchies (abstract subclasses of
    ``IFactory``). Every abstract subclass automatically gets its own
    isolated ``_impls`` dict when it is defined.

    Usage pattern::

        class IAnimalFactory(IFactory, ABC):
            pass  # gets its own registry

        class Dog(IAnimalFactory):
            @classmethod
            def registration_key(cls) -> str:
                return "dog"

        IAnimalFactory.get_registered_classes()  # {"dog"}
        IFactory.get_registered_classes()        # always empty
    """

    _impls: Dict[str, Type[Self]] = {}
    _lock = threading.Lock()

    def __init_subclass__(cls, **kwargs) -> None:
        """
        Hook called whenever ``IFactory`` is subclassed.

        - **Abstract subclass**: receives a fresh, empty ``_impls`` dict so
          it owns an isolated registry.
        - **Concrete subclass**: registered in every abstract base along its
          MRO, *excluding* ``IFactory`` itself (which is a protocol, not a
          registry).
        """

        super().__init_subclass__(**kwargs)

        if isabstract(cls):
            # Each abstract (new base) class will have its own
            # register. This way each new base interface does not need
            # to redefine `_impls`...
            cls._impls = {}

        else:
            with IFactory._lock:
                key = cls.registration_key()
                for base in type.mro(cls):
                    if (
                        base is not IFactory
                        and issubclass(base, IFactory)
                        and isabstract(base)
                        and hasattr(base, "_impls")
                    ):
                        if key in base._impls:
                            raise ValueError(
                                f"Registration key '{key}' already exists "
                                f"in {base.__name__} registry. "
                                f"Existing class: {base._impls[key].__name__}, "
                                f"Attempting to register: {cls.__name__}"
                            )

                        base._impls[key] = cls



    @classmethod
    @abstractmethod
    def registration_key(cls) -> str:
        """
        It returns the name (reference) for the key used
        to register, like: return `self.__name__`
        """




    @classmethod
    def get_registered_classes(cls) -> Set[str]:
        """It returns the current registered implementations"""
        return set(cls._impls)




    @classmethod
    def get_class(cls, cls_reference: str) -> Optional[Type[Self]]:
        """It returns the reference (class type) by reference"""
        return cls._impls.get(cls_reference, None)