-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathregistry.py
More file actions
74 lines (56 loc) · 2.72 KB
/
Copy pathregistry.py
File metadata and controls
74 lines (56 loc) · 2.72 KB
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
"""Strategy in its Python form: functions registered as interchangeable rules.
A strategy is any callable ``(argument) -> result``. ``StrategyRegistry``
collects a family of them — registering is decorating — so callers can pick
one by name, run them all, or compare their results.
"""
from __future__ import annotations
from collections.abc import Callable, Iterator
from typing import Generic, TypeVar
In_ = TypeVar("In_")
Out = TypeVar("Out")
class UnknownStrategyError(LookupError):
"""No strategy with that name is registered."""
class StrategyRegistry(Generic[In_, Out]):
"""A named family of interchangeable algorithms.
New strategies join by being defined (``@registry.register``) — the code
that *uses* the family never changes.
"""
def __init__(self) -> None:
self._strategies: dict[str, Callable[[In_], Out]] = {}
def register(
self, strategy: Callable[[In_], Out], *, replace: bool = False
) -> Callable[[In_], Out]:
"""Add a strategy under its function name; usable as a decorator.
A duplicate name is an error unless ``replace=True`` — the key is
``__name__``, so two same-named functions from different modules
collide by accident, and silently dropping a rule is how a discount
stops applying with nothing logged.
"""
name = str(getattr(strategy, "__name__", repr(strategy)))
if name in self._strategies and not replace:
raise ValueError(f"strategy {name!r} already registered (pass replace=True)")
self._strategies[name] = strategy
return strategy
def unregister(self, name: str) -> None:
"""Remove a strategy by name; membership, like order, is policy."""
try:
del self._strategies[name]
except KeyError:
known = ", ".join(sorted(self._strategies)) or "none"
raise UnknownStrategyError(f"no strategy {name!r} (known: {known})") from None
def get(self, name: str) -> Callable[[In_], Out]:
"""Look one strategy up by name."""
try:
return self._strategies[name]
except KeyError:
known = ", ".join(sorted(self._strategies)) or "none"
raise UnknownStrategyError(f"no strategy {name!r} (known: {known})") from None
def names(self) -> list[str]:
return list(self._strategies)
def results(self, argument: In_) -> dict[str, Out]:
"""Run every registered strategy on one argument, keyed by name."""
return {name: strategy(argument) for name, strategy in self._strategies.items()}
def __iter__(self) -> Iterator[Callable[[In_], Out]]:
return iter(self._strategies.values())
def __len__(self) -> int:
return len(self._strategies)