dispatch¶
Dispatch on a value the way functools.singledispatch dispatches on a type.
When to reach for it¶
Reach for dispatch when you select behavior by the value of an argument rather than its type.
Think kind="html" versus kind="text", or a message routed by its event field. The dispatch
key is the value of a keyword argument, default kind, and each implementation registers under
the value it handles.
The anti-pattern it replaces¶
It replaces the long if kind == ...: elif kind == ...: chain and the dict-of-functions you wire
up by hand. Implementations register with a decorator, the fallback handles the no-kind case, and
a clear error names the valid kinds when an unknown one arrives.
Usage¶
from patos import value_dispatch
@value_dispatch
def render(node: str, **kw: object) -> str:
"""Fallback used when no kind is given."""
return node
@render.register("html")
def render_html(node: str, **kw: object) -> str:
return f"<p>{node}</p>"
@render.register("text")
def render_text(node: str, **kw: object) -> str:
return node.upper()
render("hi", kind="html") # "<p>hi</p>"
render("hi", kind="text") # "HI"
render("hi") # "hi", the fallback
render.html("hi") # "<p>hi</p>", direct attribute call
Use the parametrized form to dispatch on a different keyword.
Public API¶
value_dispatch. Decorator. Works bare (@value_dispatch) or parametrized (@value_dispatch(kind="how"))..register(key=None, *, name=None). Return an implementation decorator. Empty parentheses infer the function name. A key ornamesets it explicitly. Identifier keys are also exposed as attributes unless they would shadow the dispatcher's own API..registry. Read-only view of the kind to implementation mapping..kinds(). All registered kinds, sorted byrepr.key in dispatcher. Membership test on registered kinds.
type_dispatch¶
The dual that keys on the type of the first positional argument rather than a keyword value,
the open-type-ladder replacement for an if isinstance(x, A): ... elif isinstance(x, B): ...
chain. Like functools.singledispatch but with the same registration ergonomics as
value_dispatch.
from patos import type_dispatch
@type_dispatch
def render(node):
return "unknown"
@render.register()
def _(node: Circle): # type read from the first parameter's annotation
return "circle"
@render.register(Square) # or given explicitly
def _(node):
return "square"
render(Circle()) # "circle"
render(object()) # "unknown" -- unmatched type falls back
type_dispatch. Decorator. Works bare (@type_dispatch) or parametrized (@type_dispatch())..register(type=None). Return an implementation decorator. Empty parentheses infer the first parameter's type annotation. An explicit type selects it directly. A call resolves the argument's type against the registry by walking its MRO, so the most specific registered base wins..registry. Read-only view of the type to implementation mapping..types(). All registered types, sorted by name.cls in dispatcher. Membership test on registered types.
Source¶
Copy this into your project and own it. No dependency, no tool, just one module you can read and change.
import functools
import inspect
from collections.abc import Callable, Hashable
from types import MappingProxyType
from typing import Protocol, TypeGuard, cast
class Named(Protocol):
"""Callable metadata needed when a registration key is inferred."""
__name__: str
class value_dispatch[**P, R]:
"""Turn a function into a value-dispatched generic.
Like `functools.singledispatch`, but the dispatch key is the *value* of a
keyword argument (default `"kind"`) rather than the *type* of the first
positional argument. Both decorator forms work:
@value_dispatch
def render(node, **kw): ... # bare: kind is "kind"
@value_dispatch(kind="how")
def render(node, **kw): ... # kind is "how"
Implementations register under a key with `@render.register()`. Each is also
exposed as an attribute (`render.html`) so callers that already know the
kind can skip the registry lookup. Calling with no kind (or `kind=None`)
falls back to the originally decorated function.
fallback: the wrapped function; supplied implicitly by the decorator.
kind: name of the keyword argument carrying the dispatch value.
"""
def __init__(
self,
fallback: Callable[P, R] | None = None,
*,
kind: str = "kind",
) -> None:
self.kind_arg = kind
self.registry_map: dict[Hashable, Callable[P, R]] = {}
self.fallback: Callable[P, R] | None = None
if fallback is not None:
self.bind_fallback(fallback)
def bind_fallback(self, fallback: Callable[P, R]) -> None:
"""Adopt `fallback` as the default impl and copy its metadata onto self.
Rejects functions whose first parameter is `self`, since a dispatcher stored in
a class body is not a descriptor and never binds as a method. `updated=()` keeps
the fallback's `__dict__` from being copied over the dispatcher's own state.
"""
if inspect.isfunction(fallback) and fallback.__code__.co_varnames[:1] == ("self",):
raise TypeError(
"value_dispatch does not support methods. Decorate a module-level "
"function instead of one whose first parameter is self.",
)
self.fallback = fallback
functools.update_wrapper(self, fallback, updated=())
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
"""Bind the fallback on the parametrised first call, otherwise dispatch.
With no fallback yet, this is the parametrised-decorator stage: the sole
argument is the function to wrap. Once bound, pop the kind keyword and
route to its registered impl, falling back when the kind is absent.
"""
if self.fallback is None:
if len(args) != 1 or kwargs or not callable(args[0]):
raise TypeError(
"this value_dispatch has no function bound yet. Apply the "
"parametrised dispatcher as a decorator to its fallback function "
"before calling it.",
)
self.bind_fallback(cast("Callable[P, R]", args[0]))
# The parametrised form `@value_dispatch(kind=...)` then `@dispatcher` returns
# the dispatcher itself, so this branch yields self rather than an `R`. The cast
# keeps both decorator stages behind one `__call__` without splitting the API.
return cast(R, self)
value = kwargs.pop(self.kind_arg, None)
if value is None:
return self.fallback(*args, **kwargs)
try:
impl = self.registry_map[value]
except KeyError:
raise ValueError(
f"{getattr(self.fallback, '__name__', 'value_dispatch')}: unknown "
f"{self.kind_arg}={value!r}; "
f"choose from {sorted(self.registry_map, key=repr)}",
) from None
return impl(*args, **kwargs)
def register[**Q](
self,
key: Hashable | None = None,
*,
name: Hashable | None = None,
) -> Callable[[Callable[Q, R]], Callable[Q, R]]:
"""Register an implementation under a kind value.
`@register()` uses the function name. `@register(key)` and
`@register(name=key)` provide an explicit key.
`Q` is its own type parameter, distinct from the dispatcher's `P`, because each
kind routinely takes its own shape (a bound `functools.partial`, an impl that
drops the fallback's catch-all `**kwargs`); only the return type `R` need agree.
`bind` is where that shape is deliberately forgotten, since dispatch keys on a
value at runtime rather than a type a checker can verify ahead of the call. When
the key is a valid identifier that does not shadow the dispatcher's own API or
state, the impl is also set as a dispatcher attribute, enabling direct
`dispatcher.foo(...)` calls. Dispatch by key works either way.
key: explicit dispatch key.
name: explicit key, taking precedence over `key`.
"""
dispatch_key = key if name is None else name
def decorate(impl: Callable[Q, R]) -> Callable[Q, R]:
return self.bind(
impl,
self.implied_key(impl) if dispatch_key is None else dispatch_key,
)
return decorate
def implied_key[**Q](self, impl: Callable[Q, R]) -> str:
"""Derive the registry key from the impl's `__name__`, failing clearly when absent."""
try:
return cast(Named, impl).__name__
except AttributeError:
raise TypeError(
f"cannot infer a kind for {impl!r} because it has no __name__. "
f"Register it with an explicit key, for example register(name=...).",
) from None
def bind[**Q](self, impl: Callable[Q, R], key: Hashable) -> Callable[Q, R]:
"""Store `impl` under `key`, also exposing it as an attribute when that is safe.
The registry only ever calls a stored impl with the arguments a matching dispatch
call actually supplies, never with `P` reconstructed from thin air, so re-typing
`impl` here to the dispatcher's own `Callable[P, R]` costs no real safety, only the
precision of `Q`, which this one boundary is exactly where that trade belongs.
"""
if self.exposable(key):
setattr(self, key, impl)
self.registry_map[key] = cast("Callable[P, R]", impl)
return impl
def exposable(self, key: Hashable) -> TypeGuard[str]:
"""Whether `key` may become a dispatcher attribute without shadowing its own API.
The key must be an identifier string colliding with neither class-level API
(`register`, `kinds`, ...) nor instance state, except for the attribute that a
previous registration of the same key already exposed.
"""
if not (isinstance(key, str) and key.isidentifier()):
return False
if hasattr(type(self), key):
return False
return key not in vars(self) or vars(self)[key] is self.registry_map.get(key)
@property
def registry(self) -> MappingProxyType[Hashable, Callable[P, R]]:
"""Read-only view of the kind to impl mapping."""
return MappingProxyType(self.registry_map)
def kinds(self) -> list[Hashable]:
"""All registered kinds, sorted by their `repr` so mixed-type keys stay orderable."""
return sorted(self.registry_map, key=repr)
def __contains__(self, key: Hashable) -> bool:
return key in self.registry_map
def __getattr__(self, name: str) -> Callable[P, R]:
"""Type the attribute `bind` exposes for an identifier-shaped key (`dispatcher.foo`).
Only reached when normal lookup misses, so a real attribute (`registry_map`, an
exposed impl `bind` has actually set) is always found first; this exists solely to
give the type checker a return type for the dynamic `dispatcher.foo(...)` surface
`register` documents, which no static attribute list can otherwise describe.
name: the exposed kind being looked up.
"""
raise AttributeError(name)
def __repr__(self) -> str:
name = getattr(self.fallback, "__name__", "(unbound)")
return f"<value_dispatch {name!r} kinds={self.kinds()}>"
class type_dispatch[**P, R]:
"""Turn a function into a type-dispatched generic, the dual of :class:`value_dispatch`.
Where `value_dispatch` keys on the *value* of a keyword argument, this keys on the *type* of
the first positional argument, the open type ladder a hand-rolled `if isinstance(x, A): ...
elif isinstance(x, B): ...` chain encodes (rendering each GPU/NPU subtype, formatting each
result kind). Like `functools.singledispatch` but registering with the same ergonomics as
`value_dispatch`:
@type_dispatch
def render(node): ... # the fallback for an unmatched type
@render.register()
def _(node: Circle): ... # type read from the first parameter's annotation
@render.register(Square)
def _(node): ... # type given explicitly
A call resolves the first argument's type against the registry by walking its MRO, so the most
specific registered base wins and a subtype falls through to a registered supertype. A type the
ladder does not cover routes to the originally decorated fallback. New cases are new
registrations, never another `elif`.
fallback: the wrapped catch-all function; supplied implicitly by the decorator.
"""
def __init__(self, fallback: Callable[P, R] | None = None) -> None:
self.registry_map: dict[type, Callable[P, R]] = {}
self.fallback: Callable[P, R] | None = None
if fallback is not None:
self.bind_fallback(fallback)
def bind_fallback(self, fallback: Callable[P, R]) -> None:
"""Adopt `fallback` as the catch-all and copy its metadata, rejecting a method fallback."""
if inspect.isfunction(fallback) and fallback.__code__.co_varnames[:1] == ("self",):
raise TypeError(
"type_dispatch does not support methods. Decorate a module-level function "
"instead of one whose first parameter is self.",
)
self.fallback = fallback
functools.update_wrapper(self, fallback, updated=())
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
"""Bind the fallback on the parametrised first call, else dispatch on `args[0]`'s type.
With no fallback yet this is the bare-decorator stage and the sole argument is the function
to wrap. Once bound, the first positional argument's type selects the implementation, and a
call with no positional argument routes to the fallback.
"""
if self.fallback is None:
if len(args) != 1 or kwargs or not callable(args[0]):
raise TypeError(
"this type_dispatch has no function bound yet. Apply it as a decorator to "
"its fallback function before calling it.",
)
self.bind_fallback(cast("Callable[P, R]", args[0]))
return cast(R, self)
impl = self.resolve(type(args[0]), self.fallback) if args else self.fallback
return impl(*args, **kwargs)
def resolve(self, cls: type, fallback: Callable[P, R]) -> Callable[P, R]:
"""The implementation for `cls`, the nearest registered base on its MRO, else `fallback`.
cls: the runtime type of the first argument.
fallback: returned when no base of `cls` is registered.
"""
for base in cls.__mro__:
if base in self.registry_map:
return self.registry_map[base]
return fallback
def register[**Q](
self,
cls: type | None = None,
) -> Callable[[Callable[Q, R]], Callable[Q, R]]:
"""Register an implementation for a type.
Two forms mirroring `value_dispatch`:
- `@register()` reads the type from the first parameter's annotation.
- `@register(SomeType)`: the type is given explicitly.
`Q` is its own type parameter, distinct from the dispatcher's `P`: an impl's first
parameter is annotated with the concrete type it handles, narrower than whatever
the fallback declares, which is the whole point of registering it. `bind` is where
that narrower shape is deliberately forgotten, the same trade `value_dispatch.bind`
makes for the same reason.
cls: explicit type, or `None` to infer it from the implementation.
"""
return lambda impl: self.bind(
self.annotated_type(impl) if cls is None else cls,
impl,
)
def annotated_type[**Q](self, impl: Callable[Q, R]) -> type:
"""The type annotation of `impl`'s first parameter, failing clearly when it is absent."""
hints = inspect.signature(impl).parameters
first = next(iter(hints.values()), None)
if first is None or first.annotation is inspect.Parameter.empty:
raise TypeError(
f"cannot infer a dispatch type for {impl!r} because its first parameter has no "
f"annotation. Register it with an explicit type, for example register(SomeType).",
)
if not isinstance(first.annotation, type):
raise TypeError(
f"the first parameter of {impl!r} is annotated {first.annotation!r}, which is not "
f"a class. Register it with an explicit type instead.",
)
return first.annotation
def bind[**Q](self, cls: type, impl: Callable[Q, R]) -> Callable[Q, R]:
"""Store `impl` as the implementation for `cls`, returning `impl` unchanged."""
self.registry_map[cls] = cast("Callable[P, R]", impl)
return impl
@property
def registry(self) -> MappingProxyType[type, Callable[P, R]]:
"""Read-only view of the type to impl mapping."""
return MappingProxyType(self.registry_map)
def types(self) -> list[type]:
"""All registered types, sorted by name so the listing is stable."""
return sorted(self.registry_map, key=lambda cls: cls.__name__)
def __contains__(self, cls: type) -> bool:
return cls in self.registry_map
def __repr__(self) -> str:
name = getattr(self.fallback, "__name__", "(unbound)")
types = [cls.__name__ for cls in self.types()]
return f"<type_dispatch {name!r} types={types}>"