effects.py

from dataclasses import dataclass
from typing import Generator, TypeVar, Any, Callable, Optional, cast, NamedTuple, Dict, Type, Tuple, Generic

R = TypeVar("R")

Eff = Generator[Tuple[Any, ...], Any, R]

def handle_op(
    g: Eff[R],
    op: Tuple[Any, ...],
    h: Dict[Type[Tuple[Any, ...]], Callable[..., Eff[R]]]
) -> Eff[R]:
    def resume(r: object = None) -> Eff[R]:
        try:
            op = g.send(r)
            # deep handler
            return (yield from handle_op(g, op, h))
        except StopIteration as e:
            return cast(R, e.value)

    # NB: No subtyping
    if type(op) in h:
        return (yield from h[type(op)](*op, resume=resume))
    else:
        r = yield op
        return (yield from resume(r))

def handler(g: Eff[R], h: Dict[Type[Tuple[Any, ...]], Callable[..., Eff[R]]]) -> Eff[R]:
    try:
        op = next(g)
        return (yield from handle_op(g, op, h))
    except StopIteration as e:
        return cast(R, e.value)

def run(g: Eff[R]) -> R:
    try:
        op = next(g)
        raise RuntimeError(f'unhandled effects {op}')
    except StopIteration as e:
        return cast(R, e.value)

class output(NamedTuple):
    arg: int

class input(NamedTuple):
    pass

def sample() -> Eff[None]:
    i = yield input()
    yield output(i)
    yield output(i + 1)
    yield output(i + 2)

def output_handler(arg: int, *, resume: Callable[[], Eff[R]]) -> Eff[R]:
    print(arg)
    # NB: this form is inefficient, because we always push a stack
    # frame even when it's not necessary
    return (yield from resume())

def input_handler(*, resume: Callable[[int], Eff[R]]) -> Eff[R]:
    return (yield from resume(3))

run(
    handler(
        sample(),
        {input: input_handler,
         output: output_handler}
    )
)

run(
    handler(
        handler(
            sample(),
            {output: output_handler}
        ),
        {input: input_handler}
    )
)

def sample2() -> Eff[int]:
    yield error()
    print("omitted")
    return 2

class error(NamedTuple):
    pass

def error_handler(*, resume: Callable[[int], Eff[R]]) -> Eff[int]:
    # ignore resume
    yield from []
    return 2

run(handler(sample2(), {error: error_handler}))