Examples

mutlimethod

Multimethods are a mapping of signatures (tuple of types) to functions. They maintain an efficient dispatch tree, and cache the called signatures.

from multimethod import multimethod
import operator

classic_div = multimethod(operator.truediv)
classic_div[int, int] = operator.floordiv
classic_div

{(): <function _operator.truediv(a, b, /)>,
 (int, int): <function _operator.floordiv(a, b, /)>}

classic_div(3, 2)

1

classic_div(3.0, 2)

1.5

classic_div

{(): <function _operator.truediv(a, b, /)>,
 (int, int): <function _operator.floordiv(a, b, /)>,
 (float, int): <function _operator.truediv(a, b, /)>}

Multimethods introspect type annotations and use the name to find existing multimethods.

import itertools
from typing import Iterable, Sequence

@multimethod
def chunks(values: Iterable, size):
    it = iter(values)
    return iter(lambda: list(itertools.islice(it, size)), [])

@multimethod
def chunks(values: Sequence, size):
    for index in range(0, len(values), size):
        yield values[index:index + size]

list(chunks(iter('abcde'), 3))

[['a', 'b', 'c'], ['d', 'e']]

list(chunks('abcde', 3))

['abc', 'de']

Multimethods also have an explicit register method similar to functools.singledispatch.

@multimethod
def window(values, size=2):
    its = itertools.tee(values, size)
    return zip(*(itertools.islice(it, index, None) for index, it in enumerate(its)))

@window.register
def _(values: Sequence, size=2):
    for index in range(len(values) - size + 1):
        yield values[index:index + size]

list(window(iter('abcde')))

[('a', 'b'), ('b', 'c'), ('c', 'd'), ('d', 'e')]

list(window('abcde'))

['ab', 'bc', 'cd', 'de']

overload

Whereas multimethods require an issubclass relationship, overloads dispatch on any predicates.

import asyncio
import time
from concurrent import futures
from multimethod import overload

@overload
def wait(timeout, func, *args):
    return futures.ThreadPoolExecutor().submit(func, *args).result(timeout)

@overload
async def wait(timeout, func: asyncio.iscoroutinefunction, *args):
    return await asyncio.wait_for(func(*args), timeout)

wait(0.5, time.sleep, 0.01)

wait(0.5, asyncio.sleep, 0.01)

<coroutine object wait at 0x7f578910bc48>

typing subscripts

Provisional support for type hints with subscripts.

import bisect
import random
from typing import Dict

@multimethod
def samples(weights: Dict):
    """Generate weighted random samples using bisection."""
    keys = list(weights)
    totals = list(itertools.accumulate(weights.values()))
    values = [total / totals[-1] for total in totals]
    while True:
        yield keys[bisect.bisect_right(values, random.random())]

@multimethod
def samples(weights: Dict[object, int]):
    """Generate weighted random samples more efficiently."""
    keys = list(itertools.chain.from_iterable([key] * weights[key] for key in weights))
    while True:
        yield random.choice(keys)

weights = {'a': 1, 'b': 2, 'c': 3}
next(samples(weights))

'c'

weights = {'a': 1.0, 'b': 2.0, 'c': 3.0}
next(samples(weights))

'c'