Introduction

pyfect is a Python port of Effect (v4). You write programs as effects: values that describe work and say, in their type, what they produce, how they can fail and which services they need. pyright infers all three, and refuses to run a program that leaves an error unhandled or a service unprovided.

Effect[A, E, R]
       │  │  └── the services it needs (Never: none)
       │  └───── the errors it can fail with (Never: none)
       └──────── the value it succeeds with

Handling an error removes it from E; providing a service removes it from R. An effect runs once both are Never:

Why#

Plain asyncio gets you concurrency. It leaves the rest to discipline: exceptions a signature doesn't mention, CancelledError that a broad except swallows, tasks that outlive the function that started them, and dependencies passed by hand or patched in tests.

pyfect handles these in the runtime and the types:

• Errors are values in the type. Effect[str, NotFound | Forbidden] says what can go wrong; handling one removes it from the type. Bugs (defects) are kept apart from expected failures.
• Services are requirements in the type. An effect that needs a Database says so, and a Layer provides it: real in production, a stub in tests.
• Structured concurrency. Fibers run on pyfect's own scheduler. A fiber's children are interrupted when it finishes, interruption can't be swallowed, and finalizers always run.
• Retries, timeouts, resources, streams, config, logs, spans and metrics are built in and compose with each other.

Compared with asyncio#

A user lookup with a timeout and retries:

"""The same program with pyfect: the errors are part of the type."""

from dataclasses import dataclass

import pyfect as fx
from pyfect import schedule


@dataclass
class UserNotFound(fx.TaggedError):
    user_id: str


def fetch_user(user_id: str) -> fx.Effect[str, UserNotFound]:
    if user_id != "1":
        return fx.sleep(0.01).flat_map(lambda _: fx.fail(UserNotFound(user_id)))
    return fx.sleep(0.01).as_("Ada")


def greet(user_id: str) -> fx.Effect[str, fx.TimeoutException]:
    return (
        fetch_user(user_id)
        .map(lambda name: f"Hello, {name}!")
        .pipe(fx.catch_tag(UserNotFound, lambda _: fx.succeed("Who?")))
        .timeout(1)
        .retry(schedule.exponential(0.1).up_to(times=2))
    )


# greet's type says what is left to handle: a timeout after the last retry.
# Running it without handling that is a type error, so we handle it here.
program = greet("1").pipe(fx.catch_tag(fx.TimeoutException, lambda _: fx.succeed("Timed out")))

assert fx.run_sync(program) == "Hello, Ada!"

The pyfect version is not shorter. What changes is that greet's type, Effect[str, TimeoutException], tells the caller exactly what it still has to handle, and pyright checks that they do.

When plain asyncio is the better choice: a small script, a codebase that does not want a new programming model, or code that mostly awaits one library. pyfect runs on asyncio (run_async, from_async), so you can adopt it one function at a time.

Status#

pyfect is alpha. It covers about 64% of Effect v4 (about 71% of the runtime core): effects, errors, fibers, layers, scopes, schedules, streams, config, logging, tracing, metrics and test utilities. Schema, STM and the platform libraries (HTTP, SQL, ...) are not done yet. PARITY.md lists what is covered, what differs and why.

Start with Installation, then Effects and generators. The API reference is generated from the code.