Examples¶

multimethod¶

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

In [1]:

from multimethod import multimethod
import operator

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

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

Out[1]:

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

In [2]:

classic_div(3, 2)

classic_div(3, 2)

Out[2]:

1

In [3]:

classic_div(3.0, 2)

classic_div(3.0, 2)

Out[3]:

1.5

In [4]:

classic_div

classic_div

Out[4]:

{(): <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.

In [5]:

import itertools
from collections.abc import Iterable, Sequence


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


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


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

import itertools from collections.abc import Iterable, Sequence @multimethod def batched(values: Iterable, size): it = iter(values) return iter(lambda: list(itertools.islice(it, size)), []) @multimethod def batched(values: Sequence, size): for index in range(0, len(values), size): yield values[index : index + size] list(batched(iter('abcde'), 3))

Out[5]:

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

In [6]:

list(batched('abcde', 3))

list(batched('abcde', 3))

Out[6]:

['abc', 'de']

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

In [7]:

@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')))

@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')))

Out[7]:

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

In [8]:

list(window('abcde'))

list(window('abcde'))

Out[8]:

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

parametric¶

In addition to issubclass, multimethods can dispatch on isinstance with parametric checks.

In [9]:

import asyncio
import time
from collections.abc import Callable
from concurrent import futures
from multimethod import parametric

Coroutine = parametric(Callable, asyncio.iscoroutinefunction)


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


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


wait(0.5, time.sleep, 0.01)

import asyncio import time from collections.abc import Callable from concurrent import futures from multimethod import parametric Coroutine = parametric(Callable, asyncio.iscoroutinefunction) @multimethod def wait(timeout, func, *args): return futures.ThreadPoolExecutor().submit(func, *args).result(timeout) @multimethod async def wait(timeout, func: Coroutine, *args): return await asyncio.wait_for(func(*args), timeout) wait(0.5, time.sleep, 0.01)

In [10]:

wait(0.5, asyncio.sleep, 0.01)

wait(0.5, asyncio.sleep, 0.01)

Out[10]:

<coroutine object wait at 0x7f8a64b32b20>

In [11]:

from array import array

IntArray = parametric(array, typecode='i')
isinstance(array('i'), IntArray)

from array import array IntArray = parametric(array, typecode='i') isinstance(array('i'), IntArray)

Out[11]:

True

In [12]:

isinstance(array('f'), IntArray)

isinstance(array('f'), IntArray)

Out[12]:

False

typing subscripts¶

Support for type hints with subscripts.

In [13]:

import bisect
import random


@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))

import bisect import random @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))

Out[13]:

'c'

In [14]:

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

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

Out[14]:

'b'