File size: 3,122 Bytes
d91766b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | from typing import Iterable
from diffulex.utils.registry import fetch_factory_name
_NOT_PROVIDED = object()
class DiffulexStrategyRegistry:
"""Registry-driven factory for module implementations."""
_DEFAULT_KEY = "__default__"
def __init_subclass__(cls, **kwargs):
"""Initialize a separate _MODULE_MAPPING for each subclass."""
super().__init_subclass__(**kwargs)
cls._MODULE_MAPPING: dict[str, object] = {}
@classmethod
def register(
cls,
strategy_name: str,
factory: object = _NOT_PROVIDED,
*,
aliases: Iterable[str] = (),
is_default: bool = False,
exist_ok: bool = False,
):
if not isinstance(strategy_name, str) or not strategy_name:
raise ValueError("strategy_name must be a non-empty string.")
if isinstance(aliases, str):
raise TypeError("aliases must be an iterable of strings, not a single string.")
def decorator(factory_fn: object):
cls._register(strategy_name, factory_fn, exist_ok=exist_ok)
for alias in dict.fromkeys(aliases):
if not isinstance(alias, str) or not alias:
raise ValueError("aliases must contain non-empty strings.")
cls._register(alias, factory_fn, exist_ok=exist_ok)
if is_default:
cls._register(cls._DEFAULT_KEY, factory_fn, exist_ok=True)
return factory_fn
if factory is _NOT_PROVIDED:
return decorator
return decorator(factory)
@classmethod
def _register(cls, key: str, factory: object, *, exist_ok: bool) -> None:
# If the same factory is already registered, silently skip (idempotent registration)
if key in cls._MODULE_MAPPING:
existing = cls._MODULE_MAPPING[key]
# Check if it's the same factory object
if existing is factory:
return # Same factory already registered, no-op
# Check if it's the same class by name and module (handles module reload cases)
existing_name = fetch_factory_name(existing)
new_name = fetch_factory_name(factory)
if existing_name == new_name:
return # Same class already registered (possibly from module reload), no-op
if not exist_ok:
raise ValueError(
f"Module '{key}: {new_name}' is already registered as '{existing_name}'. "
f"Use exist_ok=True to override."
)
cls._MODULE_MAPPING[key] = factory
@classmethod
def unregister(cls, strategy_name: str) -> None:
cls._MODULE_MAPPING.pop(strategy_name, None)
@classmethod
def available_modules(cls) -> tuple[str, ...]:
return tuple(sorted(k for k in cls._MODULE_MAPPING if k != cls._DEFAULT_KEY))
@classmethod
def _ensure_strategies_loaded(cls) -> None:
"""Ensure strategy modules are imported to trigger registration."""
if not cls._MODULE_MAPPING:
from diffulex import strategy as _ # noqa: F401
|