Examples¶
mutlimethod¶
Multimethods are a mapping of signatures (tuple of types) to functions. They maintain an efficient dispatch tree, and cache the called signatures.
[2]:
from multimethod import multimethod
import operator
classic_div = multimethod(operator.truediv)
classic_div[int, int] = operator.floordiv
classic_div
[2]:
{(): <function _operator.truediv(a, b, /)>,
(int, int): <function _operator.floordiv(a, b, /)>}
[3]:
classic_div(3, 2)
[3]:
1
[4]:
classic_div(3.0, 2)
[4]:
1.5
[5]:
classic_div
[5]:
{(): <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.
[6]:
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))
[6]:
[['a', 'b', 'c'], ['d', 'e']]
[7]:
list(chunks('abcde', 3))
[7]:
['abc', 'de']
Multimethods also have an explicit register
method similar to functools.singledispatch
.
[8]:
@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')))
[8]:
[('a', 'b'), ('b', 'c'), ('c', 'd'), ('d', 'e')]
[9]:
list(window('abcde'))
[9]:
['ab', 'bc', 'cd', 'de']
overload¶
Whereas multimethods require an issubclass
relationship, overloads dispatch on any predicates.
[10]:
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)
[11]:
wait(0.5, asyncio.sleep, 0.01)
[11]:
<coroutine object wait at 0x7f4654526c48>
typing subscripts¶
Provisional support for type hints with subscripts.
[12]:
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))
[12]:
'a'
[13]:
weights = {'a': 1.0, 'b': 2.0, 'c': 3.0}
next(samples(weights))
[13]:
'a'