Reference

multimethod.multimethod

Bases: dict

A callable directed acyclic graph of methods.

Source code in multimethod/__init__.py

class multimethod(dict):
    """A callable directed acyclic graph of methods."""

    __name__: str
    pending: set
    generics: list[tuple]  # positional bases which require instance checks

    def __new__(cls, func):
        homonym = inspect.currentframe().f_back.f_locals.get(func.__name__)
        if isinstance(homonym, multimethod):
            return homonym

        self = functools.update_wrapper(dict.__new__(cls), func)
        self.pending = set()
        self.generics = []
        return self

    def __init__(self, func: Callable):
        try:
            self[signature.from_hints(func)] = func
        except (NameError, AttributeError):
            self.pending.add(func)

    @tp_overload
    def register(self, __func: REGISTERED) -> REGISTERED: ...  # pragma: no cover

    @tp_overload
    def register(self, *args: type) -> Callable[[REGISTERED], REGISTERED]: ...  # pragma: no cover

    def register(self, *args) -> Callable:
        """Decorator for registering a function.

        Optionally call with types to return a decorator for unannotated functions.
        """
        if len(args) == 1 and hasattr(args[0], '__annotations__'):
            multimethod.__init__(self, *args)
            return self if self.__name__ == args[0].__name__ else args[0]
        return lambda func: self.__setitem__(args, func) or func

    def __get__(self, instance, owner):
        return self if instance is None else types.MethodType(self, instance)

    def parents(self, types: tuple) -> set:
        """Find immediate parents of potential key."""
        parents = {key for key in list(self) if isinstance(key, signature) and key < types}
        return parents - {ancestor for parent in parents for ancestor in parent.parents}

    def clean(self):
        """Empty the cache."""
        for key in list(self):
            if not isinstance(key, signature):
                super().__delitem__(key)

    def copy(self):
        """Return a new multimethod with the same methods."""
        return dict.__new__(type(self)).__ior__(self)

    def __setitem__(self, types: tuple, func: Callable):
        self.clean()
        if not isinstance(types, signature):
            types = signature(types)
        parents = types.parents = self.parents(types)
        with contextlib.suppress(ValueError):
            types.sig = inspect.signature(func)
        self.pop(types, None)  # ensure key is overwritten
        for key in self:
            if types < key and (not parents or parents & key.parents):
                key.parents -= parents
                key.parents.add(types)
        for index, cls in enumerate(types):
            if origins := set(subtype.origins(cls)):
                self.generics += [()] * (index + 1 - len(self.generics))
                self.generics[index] = tuple(origins.union(self.generics[index]))
        super().__setitem__(types, func)
        self.__doc__ = self.docstring

    def __delitem__(self, types: tuple):
        self.clean()
        super().__delitem__(types)
        for key in self:
            if types in key.parents:
                key.parents = self.parents(key)
        self.__doc__ = self.docstring

    def select(self, types: tuple, keys: set[signature]) -> Callable:
        keys = {key for key in keys if key.callable(*types)}
        funcs = {self[key] for key in keys}
        if len(funcs) > 1:
            groups = collections.defaultdict(set)
            for key in keys:
                groups[types - key].add(key)
            keys = groups[min(groups)]
            funcs = {self[key] for key in keys}
            if len(funcs) == 1:
                warnings.warn("positional distance tie-breaking is deprecated", DeprecationWarning)
        if len(funcs) == 1:
            return funcs.pop()
        raise DispatchError(f"{self.__name__}: {len(keys)} methods found", types, keys)

    def __missing__(self, types: tuple) -> Callable:
        """Find and cache the next applicable method of given types."""
        self.evaluate()
        types = tuple(map(subtype, types))
        if types in self:
            return self[types]
        return self.setdefault(types, self.select(types, self.parents(types)))

    def dispatch(self, *args) -> Callable:
        types = tuple(map(type, args))
        if not any(map(issubclass, types, self.generics)):
            return self[types]
        matches = {key for key in list(self) if isinstance(key, signature) and key.instances(*args)}
        matches -= {ancestor for match in matches for ancestor in match.parents}
        return self.select(types, matches)

    def __call__(self, *args, **kwargs):
        """Resolve and dispatch to best method."""
        self.evaluate()
        func = self.dispatch(*args)
        try:
            return func(*args, **kwargs)
        except TypeError as ex:
            raise DispatchError(f"Function {func.__code__}") from ex

    def evaluate(self):
        """Evaluate any pending forward references."""
        while self.pending:
            func = self.pending.pop()
            self[signature.from_hints(func)] = func

    @property
    def docstring(self):
        """a descriptive docstring of all registered functions"""
        docs = []
        for key, func in self.items():
            sig = getattr(key, 'sig', '')
            if func.__doc__:
                docs.append(f'{func.__name__}{sig}\n    {func.__doc__}')
        return '\n\n'.join(docs)

docstring property

a descriptive docstring of all registered functions

__call__(*args, **kwargs)

Resolve and dispatch to best method.

Source code in multimethod/__init__.py

def __call__(self, *args, **kwargs):
    """Resolve and dispatch to best method."""
    self.evaluate()
    func = self.dispatch(*args)
    try:
        return func(*args, **kwargs)
    except TypeError as ex:
        raise DispatchError(f"Function {func.__code__}") from ex

__missing__(types)

Find and cache the next applicable method of given types.

Source code in multimethod/__init__.py

def __missing__(self, types: tuple) -> Callable:
    """Find and cache the next applicable method of given types."""
    self.evaluate()
    types = tuple(map(subtype, types))
    if types in self:
        return self[types]
    return self.setdefault(types, self.select(types, self.parents(types)))

clean()

Empty the cache.

Source code in multimethod/__init__.py

def clean(self):
    """Empty the cache."""
    for key in list(self):
        if not isinstance(key, signature):
            super().__delitem__(key)

copy()

Return a new multimethod with the same methods.

Source code in multimethod/__init__.py

def copy(self):
    """Return a new multimethod with the same methods."""
    return dict.__new__(type(self)).__ior__(self)

evaluate()

Evaluate any pending forward references.

Source code in multimethod/__init__.py

def evaluate(self):
    """Evaluate any pending forward references."""
    while self.pending:
        func = self.pending.pop()
        self[signature.from_hints(func)] = func

parents(types)

Find immediate parents of potential key.

Source code in multimethod/__init__.py

def parents(self, types: tuple) -> set:
    """Find immediate parents of potential key."""
    parents = {key for key in list(self) if isinstance(key, signature) and key < types}
    return parents - {ancestor for parent in parents for ancestor in parent.parents}

register(*args)

Decorator for registering a function.

Optionally call with types to return a decorator for unannotated functions.

Source code in multimethod/__init__.py

def register(self, *args) -> Callable:
    """Decorator for registering a function.

    Optionally call with types to return a decorator for unannotated functions.
    """
    if len(args) == 1 and hasattr(args[0], '__annotations__'):
        multimethod.__init__(self, *args)
        return self if self.__name__ == args[0].__name__ else args[0]
    return lambda func: self.__setitem__(args, func) or func

multimethod.multidispatch

Bases: multimethod, dict[tuple[type, ...], Callable[..., RETURN]]

Wrapper for compatibility with functools.singledispatch.

Only uses the register method instead of namespace lookup. Allows dispatching on keyword arguments based on the first function signature.

Source code in multimethod/__init__.py

class multidispatch(multimethod, dict[tuple[type, ...], Callable[..., RETURN]]):
    """Wrapper for compatibility with `functools.singledispatch`.

    Only uses the [register][multimethod.multimethod.register] method instead of namespace lookup.
    Allows dispatching on keyword arguments based on the first function signature.
    """

    signature: Optional[inspect.Signature]

    def __new__(cls, func: Callable[..., RETURN]) -> "multidispatch[RETURN]":
        return functools.update_wrapper(dict.__new__(cls), func)

    def __init__(self, func: Callable[..., RETURN]) -> None:
        self.pending = set()
        self.generics = []
        try:
            self.signature = inspect.signature(func)
        except ValueError:
            self.signature = None
        msg = "base implementation will eventually ignore annotations as `singledispatch does`"
        with contextlib.suppress(NameError, AttributeError, TypeError):
            hints = signature.from_hints(func)
            if hints and all(map(issubclass, hints, hints)):
                warnings.warn(msg, DeprecationWarning)
        super().__init__(func)

    def __get__(self, instance, owner) -> Callable[..., RETURN]:
        return self if instance is None else types.MethodType(self, instance)  # type: ignore

    def __call__(self, *args: Any, **kwargs: Any) -> RETURN:
        """Resolve and dispatch to best method."""
        params = self.signature.bind(*args, **kwargs).args if (kwargs and self.signature) else args
        func = self.dispatch(*params)
        return func(*args, **kwargs)

__call__(*args, **kwargs)

Resolve and dispatch to best method.

Source code in multimethod/__init__.py

def __call__(self, *args: Any, **kwargs: Any) -> RETURN:
    """Resolve and dispatch to best method."""
    params = self.signature.bind(*args, **kwargs).args if (kwargs and self.signature) else args
    func = self.dispatch(*params)
    return func(*args, **kwargs)

multimethod.subtype

Bases: ABCMeta

A normalized generic type which checks subscripts.

Transforms a generic alias into a concrete type which supports issubclass and isinstance. If the type ends up being equivalent to a builtin, the builtin is returned.

Source code in multimethod/__init__.py

class subtype(abc.ABCMeta):
    """A normalized generic type which checks subscripts.

    Transforms a generic alias into a concrete type which supports `issubclass` and `isinstance`.
    If the type ends up being equivalent to a builtin, the builtin is returned.
    """

    __origin__: type
    __args__: tuple

    def __new__(cls, tp, *args):
        if tp is Any:
            return object
        if hasattr(tp, '__supertype__'):  # isinstance(..., NewType) only supported >=3.10
            tp = tp.__supertype__
        if isinstance(tp, TypeVar):
            if not tp.__constraints__:
                return object
            tp = Union[tp.__constraints__]
        origin = get_origin(tp) or tp
        if hasattr(types, 'UnionType') and isinstance(tp, types.UnionType):
            origin = Union  # `|` syntax added in 3.10
        args = tuple(map(cls, get_args(tp) or args))
        if set(args) <= {object} and not (origin is tuple and args):
            return origin
        bases = (origin,) if type(origin) in (type, abc.ABCMeta) else ()
        if origin is Literal:
            bases = (subtype(Union[tuple(map(type, args))]),)
        if origin is Union:
            counts = collections.Counter()
            for arg in args:
                counts.update(cls for cls in get_mro(arg) if issubclass(abc.ABCMeta, type(cls)))
            bases = tuple(cls for cls in counts if counts[cls] == len(args))[:1]
        if origin is Callable and args[:1] == (...,):
            args = args[1:]
        namespace = {'__origin__': origin, '__args__': args}
        return type.__new__(cls, str(tp), bases, namespace)

    def __init__(self, tp, *args): ...

    def key(self) -> tuple:
        return self.__origin__, *self.__args__

    def __eq__(self, other) -> bool:
        return hasattr(other, '__origin__') and self.key() == subtype.key(other)

    def __hash__(self) -> int:
        return hash(self.key())

    def __subclasscheck__(self, subclass):
        origin = get_origin(subclass) or subclass
        args = get_args(subclass)
        if origin is Literal:
            return all(isinstance(arg, self) for arg in args)
        if origin is Union:
            return all(issubclass(cls, self) for cls in args)
        if self.__origin__ is Literal:
            return False
        if self.__origin__ is Union:
            return issubclass(subclass, self.__args__)
        if self.__origin__ is Callable:
            return (
                origin is Callable
                and signature(self.__args__[-1:]) <= signature(args[-1:])  # covariant return
                and signature(args[:-1]) <= signature(self.__args__[:-1])  # contravariant args
            )
        return (  # check args first to avoid recursion error: python/cpython#73407
            len(args) == len(self.__args__)
            and issubclass(origin, self.__origin__)
            and all(pair[0] is pair[1] or issubclass(*pair) for pair in zip(args, self.__args__))
        )

    def __instancecheck__(self, instance):
        if self.__origin__ is Literal:
            return any(type(arg) == type(instance) and arg == instance for arg in self.__args__)
        if self.__origin__ is Union:
            return isinstance(instance, self.__args__)
        if hasattr(instance, '__orig_class__'):  # user-defined generic type
            return issubclass(instance.__orig_class__, self)
        if self.__origin__ is type:  # a class argument is expected
            return inspect.isclass(instance) and issubclass(instance, self.__args__)
        if not isinstance(instance, self.__origin__) or isinstance(instance, Iterator):
            return False
        if self.__origin__ is Callable:
            return issubclass(subtype(Callable, *get_type_hints(instance).values()), self)
        if self.__origin__ is tuple and self.__args__[-1:] != (...,):
            if len(instance) != len(self.__args__):
                return False
        elif issubclass(self, Mapping):
            instance = next(iter(instance.items()), ())
        else:
            instance = itertools.islice(instance, 1)
        return all(map(isinstance, instance, self.__args__))

    @functools.singledispatch
    def origins(self) -> Iterable[type]:
        """Return origin types which would require instance checks.

        Provisional custom usage: `subtype.origins.register(<metaclass>, lambda cls: ...)
        """
        origin = get_origin(self)
        if origin is Literal:
            yield from set(map(type, self.__args__))
        elif origin is Union:
            for arg in self.__args__:
                yield from subtype.origins(arg)
        elif origin is not None:
            yield origin

origins()

Return origin types which would require instance checks.

Provisional custom usage: `subtype.origins.register(, lambda cls: ...)

Source code in multimethod/__init__.py

@functools.singledispatch
def origins(self) -> Iterable[type]:
    """Return origin types which would require instance checks.

    Provisional custom usage: `subtype.origins.register(<metaclass>, lambda cls: ...)
    """
    origin = get_origin(self)
    if origin is Literal:
        yield from set(map(type, self.__args__))
    elif origin is Union:
        for arg in self.__args__:
            yield from subtype.origins(arg)
    elif origin is not None:
        yield origin

multimethod.parametric

Bases: ABCMeta

A type which further customizes issubclass and isinstance beyond the base type.

Parameters:

Name	Type	Description	Default
base		base type	required
funcs		all predicate functions are checked against the instance	required
attrs		all attributes are checked for equality	required

Source code in multimethod/__init__.py

class parametric(abc.ABCMeta):
    """A type which further customizes `issubclass` and `isinstance` beyond the base type.

    Args:
        base: base type
        funcs: all predicate functions are checked against the instance
        attrs: all attributes are checked for equality
    """

    def __new__(cls, base: type, *funcs: Callable, **attrs):
        return super().__new__(cls, base.__name__, (base,), {'funcs': funcs, 'attrs': attrs})

    def __init__(self, *_, **__): ...

    def __subclasscheck__(self, subclass):
        missing = object()
        attrs = getattr(subclass, 'attrs', {})
        return (
            set(subclass.__bases__).issuperset(self.__bases__)  # python/cpython#73407
            and set(getattr(subclass, 'funcs', ())).issuperset(self.funcs)
            and all(attrs.get(name, missing) == self.attrs[name] for name in self.attrs)
        )

    def __instancecheck__(self, instance):
        missing = object()
        return (
            isinstance(instance, self.__bases__)
            and all(func(instance) for func in self.funcs)
            and all(getattr(instance, name, missing) == self.attrs[name] for name in self.attrs)
        )

    def __and__(self, other):
        (base,) = set(self.__bases__ + other.__bases__)
        return type(self)(base, *set(self.funcs + other.funcs), **(self.attrs | other.attrs))

multimethod.multimeta

Bases: type

Convert all callables in namespace to multimethods.

Source code in multimethod/__init__.py

class multimeta(type):
    """Convert all callables in namespace to multimethods."""

    class __prepare__(dict):
        def __init__(*args):
            pass

        def __setitem__(self, key, value):
            if callable(value):
                value = getattr(self.get(key), 'register', multimethod)(value)
            super().__setitem__(key, value)

multimethod.DispatchError

Bases: TypeError

Source code in multimethod/__init__.py

class DispatchError(TypeError):
    pass