65 lines
1.5 KiB
Python
65 lines
1.5 KiB
Python
"""Registry for selectors and allocators — pluggable extension points."""
|
|
|
|
_selectors = {}
|
|
_allocators = {}
|
|
|
|
|
|
class RegistryError(Exception):
|
|
pass
|
|
|
|
|
|
class SelectorNotFoundError(RegistryError):
|
|
pass
|
|
|
|
|
|
class AllocatorNotFoundError(RegistryError):
|
|
pass
|
|
|
|
|
|
def register_selector(type_name):
|
|
def decorator(cls):
|
|
_selectors[type_name] = cls
|
|
cls.type = type_name
|
|
return cls
|
|
return decorator
|
|
|
|
|
|
def get_selector(spec):
|
|
if not spec or not isinstance(spec, dict):
|
|
raise ValueError(f"Invalid selector spec: {spec!r}")
|
|
type_name = spec.get("type")
|
|
if type_name not in _selectors:
|
|
raise SelectorNotFoundError(
|
|
f"No selector registered for type {type_name!r}. "
|
|
f"Available: {list(_selectors.keys())}"
|
|
)
|
|
return _selectors[type_name](spec)
|
|
|
|
|
|
def register_allocator(strategy_name):
|
|
def decorator(cls):
|
|
_allocators[strategy_name] = cls
|
|
cls.strategy = strategy_name
|
|
return cls
|
|
return decorator
|
|
|
|
|
|
def get_allocator(spec):
|
|
if not spec or not isinstance(spec, dict):
|
|
raise ValueError(f"Invalid allocation spec: {spec!r}")
|
|
strategy = spec.get("strategy")
|
|
if strategy not in _allocators:
|
|
raise AllocatorNotFoundError(
|
|
f"No allocator registered for strategy {strategy!r}. "
|
|
f"Available: {list(_allocators.keys())}"
|
|
)
|
|
return _allocators[strategy](spec)
|
|
|
|
|
|
def registered_selectors():
|
|
return dict(_selectors)
|
|
|
|
|
|
def registered_allocators():
|
|
return dict(_allocators)
|