Source code for app.engine.strategy_registry
"""Explicit strategy registry.
Replaces the legacy filesystem-scan in
:mod:`app.engine.all_engine_strategies`. Strategy classes opt-in by
decorating themselves with :func:`register_strategy`. External packages
can contribute strategies via the ``game.strategies`` entry-point group
declared in their packaging metadata.
"""
from __future__ import annotations
import logging
from importlib import metadata
from typing import Callable, TypeVar
_log = logging.getLogger(__name__)
T = TypeVar("T", bound=type)
_REGISTRY: dict[str, type] = {}
_external_loaded: bool = False
[docs]
def register_strategy(id: str, *, version: str | None = None) -> Callable[[T], T]:
"""Register a strategy class under a stable, public id.
The id is the value persisted on games and returned by the API, so it must
remain stable across class renames or file moves.
Args:
id: Public identifier for the strategy (e.g. ``"default"``).
version: Optional version string. When provided, it is also exposed as
the ``__strategy_version__`` class attribute.
Raises:
ValueError: If ``id`` is empty or another class is already registered
under the same id.
"""
if not id or not isinstance(id, str):
raise ValueError("Strategy id must be a non-empty string")
def decorator(cls: T) -> T:
"""Stamp the id/version onto ``cls`` and add it to the registry."""
existing = _REGISTRY.get(id)
if existing is not None and existing is not cls:
raise ValueError(
f"Strategy id {id!r} already registered by "
f"{existing.__module__}.{existing.__name__}"
)
cls.__strategy_id__ = id
if version is not None:
cls.__strategy_version__ = version
_REGISTRY[id] = cls
return cls
return decorator
def _load_external_strategies() -> None:
"""Import third-party strategies via the ``game.strategies`` entry point.
Each entry point should resolve to a strategy class (or to its module);
importing it triggers the :func:`register_strategy` decorator.
"""
global _external_loaded
if _external_loaded:
return
_external_loaded = True
try:
eps = metadata.entry_points(group="game.strategies")
except TypeError:
eps = metadata.entry_points().get("game.strategies", [])
for ep in eps:
try:
ep.load()
except Exception as exc:
_log.warning("Failed loading strategy entry point %s: %s", ep, exc)
[docs]
def registered_strategies() -> dict[str, type]:
"""Return a snapshot of the registry as ``{id: class}``."""
_load_external_strategies()
return dict(_REGISTRY)
[docs]
def get_registered_class(strategy_id: str) -> type | None:
"""Lookup a registered strategy class by id, or ``None`` if missing."""
_load_external_strategies()
return _REGISTRY.get(strategy_id)
[docs]
def clear_registry() -> None:
"""Reset the registry. Intended for tests only."""
global _external_loaded
_REGISTRY.clear()
_external_loaded = False