Effects & Handlers

def effect

effect(name: str, /) → Effect[..., Any]

effect( *functions: (...) → Any, ) → (P → R) → P → R

Create an effect or an effect-set decorator.

effect("name") — create a new Effect with the given name.

# Create effects
read = effect("read")
write = effect("write")

@effect(e1, e2, ...) — decorate a function to declare which effects it uses. The decorated function gains an __effects__ attribute (a frozenset of effects) that can be retrieved with effects(fn).

# Use as decorator to declare effect dependencies
@effect(read, write)
def process():
    s = read()
    return write(s)

class Effect

class Effect[**P, R]

Bases: Protocol, Generic[P, R]

An algebraic effect declaration.

Effects are created via the effect() factory and invoked like regular function calls. A handler intercepts these calls and provides the implementation at runtime.

P is the parameter types and R is the return type of the effect.

# effect: () -> str
read: Effect[[], str] = effect("read")

# effect: str -> int
write: Effect[[str], int] = effect("write")

# Inside a handler, effects are called like regular functions:
x = read()  # :: str
n = write("data")  # :: int

property name: str: The name of the effect, as passed to effect("name").

class Resume

class Resume[R, V]

Bases: Protocol, Generic

Continuation passed to synchronous effect handlers.

R is the type of the value passed to the continuation, V is the return type of the handled computation.

Calling k(value) resumes the suspended computation with value and drives it to completion (or to the next effect). The return value of k() is the final result of the handled computation.

@h.on(read)
def _read(k: Resume[str, int]) -> int:
    # k :: str -> int
    n = k("read from file")   # resume with "read from file", get the final result
    return n

The handler may call k zero times (abort), once (one-shot), or multiple times (multi-shot).

class ResumeAsync

class ResumeAsync[R, V]

Bases: Protocol, Generic

Continuation passed to asynchronous effect handlers.

R is the type of the value passed to the continuation, V is the return type of the handled computation.

Same semantics as Resume, but await k(value) is required because the handler function is async def.

@h.on(read)
async def _read(k: ResumeAsync[str, int]) -> int:
    # k :: str -> Awaitable[int]
    n = await k("read from file")
    return n

class Handler

class Handler[V]

Bases: Protocol, Generic

Protocol for synchronous effect handlers.

Create instances via create_handler. Register effect implementations with on, then invoke the handler with a caller function:

h: Handler[str] = create_handler(read, write)
# h: () -> str

@h.on(read)
def _read(k: Resume[str, str]) -> str:
    return k("data")

result = h(lambda: read())  # :: str

property effects: frozenset[Effect[..., Any]]: The effects declared for this handler.

property shallow: bool: Whether this is a shallow handler.

on( effect: Effect[P, R], ) → (EffectHandler[P, V, R]) → EffectHandler[P, V, R]: Register a handler function for effect. Returns a decorator.

check(caller: () → V) → None: Raise ValueError if any declared effect has no registered handler.

__call__( caller: () → V, *, check: bool = True, ) → V: Run caller with the registered effect handlers active.

class AsyncHandler

class AsyncHandler[V]

Bases: Protocol, Generic

Protocol for asynchronous effect handlers.

Create instances via create_async_handler. Handler functions are async def and receive a ResumeAsync continuation:

h: AsyncHandler[str] = create_async_handler(read)
# h: () -> Awaitable[str]

@h.on(read)
async def _read(k: ResumeAsync[str, str]) -> str:
    return await k("data")

result = await h(lambda: read())  # :: str

property effects: frozenset[Effect[..., Any]]: The effects declared for this handler.

property shallow: bool: Whether this is a shallow handler.

on( effect: Effect[P, R], ) → (AsyncEffectHandler[P, V, R]) → AsyncEffectHandler[P, V, R]: Register an async handler function for effect. Returns a decorator.

check( caller: () → V | Coroutine[Any, Any, V], ) → None: Raise ValueError if any declared effect has no registered handler.

async __call__( caller: () → V | Coroutine[Any, Any, V], *, check: bool = True, ) → V: Run caller with the registered async effect handlers active.

def create_handler

create_handler( *effects: Effect[..., Any], shallow: bool = False, ) → Handler[Any]

Create a synchronous handler that handles the given effects.

Register implementations with on, then call the handler with a caller function to run the computation.

If shallow is True, the handler is removed from the handler stack after handling one effect occurrence. Subsequent occurrences of the same effect will not be caught by this handler.

read: Effect[[], str] = effect("read")
h: Handler[str] = create_handler(read)

@h.on(read)
def _read(k: Resume[str, str]):
    return k("hello")

result = h(lambda: read() + " world")
# result == "hello world"

def create_async_handler

create_async_handler( *effects: Effect[..., Any], shallow: bool = False, ) → AsyncHandler[Any]

Create an asynchronous handler that handles the given effects.

Handler functions are async def and receive ResumeAsync. The caller function runs in a greenlet; effect invocations are synchronous from the caller’s perspective.

If shallow is True, the handler is removed from the handler stack after handling one effect occurrence.

read: Effect[[], str] = effect("read")
h: AsyncHandler[str] = create_async_handler(read)

@h.on(read)
async def _read(k: ResumeAsync[str, str]):
    return await k("hello")

result = await h(lambda: read() + " world")
# result == "hello world"

class EffectNotHandledError

class EffectNotHandledError[**P, R]

Bases: RuntimeError, Generic[P, R]

Raised when an effect is invoked but no handler is active for it.

def effects

effects( fn: (...) → Any, ) → frozenset[Effect[..., Any]]

Return the set of effects used by the given function.

The function must have been decorated with @effect(e1, e2, ...).

read: Effect[[], str] = effect("read")
write: Effect[[str], int] = effect("write")

@effect(read, write)
def process():
    s = read()
    return write(s)

effects(process)  # frozenset({read, write})

def unhandled_effects

unhandled_effects( fn: (...) → Any, *handlers: Handler | AsyncHandler, ) → frozenset[Effect[..., Any]]

Return the set of effects used by fn that are not handled by handlers.

read: Effect[[], str] = effect("read")
write: Effect[[str], int] = effect("write")

@effect(read, write)
def process():
    ...

h = create_handler(read)

@h.on(read)
def _read(k):
    return k("")

unhandled_effects(process, h)  # frozenset({write})