API Reference

Effect

E.EffectonError

@dataclass(frozen=True)
class EffectonError(Exception):
    ...

E.Fail

@final
@dataclass(frozen=True)
class Fail[E: EffectonError]:
    error: E

E.Die

@final
@dataclass(frozen=True)
class Die:
    defect: Any

E.Interrupt

@final
@dataclass(frozen=True)
class Interrupt:
    exception: BaseException

The effect was cut short by a cancellation.

Carries the BaseException that signalled it, such as asyncio.CancelledError. The runner consumes the cancellation and settles as Failure(Interrupt(exception)); a caller whose task should stop re-raises the exception. Like Die, it is not an error: catch_all and catch only handle Fail.

E.Cause

type Cause[E: EffectonError] = Fail[E] | Die | Interrupt

E.Effect

class Effect[A, E: EffectonError = Never, R = Never]:
    ...

E.Effect.flat_map

def flat_map[B, E2: EffectonError, R2](
    f: Callable[[A], Effect[B, E2, R2]],
) -> Effect[B, E | E2, R | R2]

E.Effect.map

def map[B](f: Callable[[A], B]) -> Effect[B, E, R]

E.Effect.catch_all

def catch_all[B, E2: EffectonError, R2](
    f: Callable[[E], Effect[B, E2, R2]],
) -> Effect[A | B, E2, R | R2]

E.Effect.catch

def catch[T: EffectonError](error_type: type[T]) -> CatchBinder[A, E, R, T]

E.Effect.on_exit

@overload
def on_exit[R2](finalizer: Effect[Any, Never, R2]) -> Effect[A, E, R | R2]
 
@overload
def on_exit[R2](
    finalizer: Callable[[Succeeded[A] | Failure[E]], Effect[Any, Never, R2]],
) -> Effect[A, E, R | R2]

Run a finalizer once this effect settles, on any outcome.

The finalizer is an effect, or a function receiving the Exit (Succeeded, or Failure carrying Fail, Die or Interrupt) that returns one. A finalizer that fails or raises replaces the outcome with its defect.

E.Effect.provide

def provide[T](requirement_type: TypeForm[T]) -> ProvideBinder[A, E, R, T]

E.Effect.scoped

def scoped[A2, E2: EffectonError, R2 = Never](
    self: Effect[A2, E2, Scope | R2],
) -> Effect[A2, E2, R2]

E.Effect.timeout

def timeout(
    duration: timedelta | None = None,
    **parts: Unpack[Parts],
) -> Effect[A, E | TimeoutException, R]

Fail with TimeoutException after a timedelta or its parts (seconds=...).

E.Effect.retry

def retry(
    schedule: Schedule | None = None,
    *,
    times: int | None = None,
    until: Callable[[E], bool] | None = None,
) -> Effect[A, E, R]

Re-run on failure while the schedule recurs, at most times more times.

E.Effect.with_span

def with_span(
    name: str,
    *,
    kind: SpanKind = 'internal',
    **attributes: object,
) -> Effect[A, E, R]

E.success

def success[A](value: A) -> Effect[A]

E.sync

def sync[A](fn: Callable[[], A]) -> Effect[A]

E.coroutine

def coroutine[A](fn: Callable[[], Awaitable[A]]) -> Effect[A]

Defer an awaitable; run_main and the run_async family can interpret it.

The thunk runs once per run of the effect and must build a fresh awaitable each time, because a coroutine object can be awaited only once. Every exception, whether raised by the thunk or by the await, becomes a defect; use attempt_async for typed failures. Interpreting the effect with run_sync_exit settles as Die(AsyncEffectInSyncRun()); run_sync raises it.

E.fail

def fail[E: EffectonError](error: E) -> Effect[Never, E]

E.die

def die(defect: Any) -> Effect[Never]

E.require

def require[R](requirement_type: TypeForm[R]) -> Effect[R, Never, R]

ProvideBinder

@final
@dataclass(frozen=True)
class ProvideBinder(Generic[A, E, R, T]):
    effect: Effect[A, E, R]
    requirement_type: TypeForm[T]

Not exported as E.ProvideBinder; reached through the return types above.

One step of effect.provide(T)(impl): T is bound, impl pending.

Calling it subtracts T from the effect's R and returns the effect with the remaining requirements.

ProvideBinder.call

def __call__[A2, E2: EffectonError, R2 = Never](
    self: ProvideBinder[A2, E2, T | R2, T],
    impl: T,
) -> Effect[A2, E2, R2]

CatchBinder

@final
@dataclass(frozen=True)
class CatchBinder(Generic[A, E, R, T]):
    effect: Effect[A, E, R]
    error_type: type[T]

Not exported as E.CatchBinder; reached through the return types above.

One step of effect.catch(T)(handler): T is bound, handler pending.

Catch particular error E, return new Effect.

CatchBinder.call

def __call__[A2, B, E3: EffectonError, R2, R3, E2: EffectonError = Never](
    self: CatchBinder[A2, T | E2, R2, T],
    handler: Callable[[T], Effect[B, E3, R3]],
) -> Effect[A2 | B, E2 | E3, R2 | R3]

Constructing effects

E.EffectGen

type EffectGen[A, E: EffectonError = Never, R = Never] = Generator[Effect[Any, E, R], Any, A]

Return type for @gen generator functions: yields effects, returns A.

E.gen

def gen[**P, A2, E2: EffectonError, R2](
    f: Callable[P, EffectGen[A2, E2, R2]],
) -> Callable[P, Effect[A2, E2, R2]]

Turn a generator function into a factory of effects.

The interpreter runs each yielded effect and sends its result back; the generator's return value becomes the effect's success value. Use x = yield from effect instead of x = yield effect so the type of x is inferred correctly. The error and requirement channels flow from the annotated yield type either way.

Each run creates a fresh generator, so the returned effect is a reusable value. A failing yielded effect abandons the generator: try/except around a yield never observes effect failures (use catch_all), and finally blocks run only when the abandoned generator is garbage collected.

E.suspend

@overload
def suspend[A, E: EffectonError, R](f: Callable[[], Effect[A, E, R]]) -> Effect[A, E, R]
 
@overload
def suspend[**P, A, E: EffectonError, R](
    f: Callable[P, Effect[A, E, R]],
) -> Callable[P, Effect[A, E, R]]

E.attempt

def attempt[A, E: EffectonError](
    thunk: Callable[[], A],
    on_error: Callable[[Exception], E],
) -> Effect[A, E]

Run an exception-throwing thunk lazily, with typed failures.

The thunk runs once per run of the effect, like sync. When it raises, on_error maps the exception into the typed error channel — under sync, every exception becomes an uncatchable defect. To keep an unexpected exception a defect, re-raise it from on_error.

E.attempt_async

def attempt_async[A, E: EffectonError](
    thunk: Callable[[], Awaitable[A]],
    on_error: Callable[[Exception], E],
) -> Effect[A, E]

Await an exception-throwing thunk lazily, with typed failures.

The async counterpart of attempt: the thunk builds a fresh awaitable once per run of the effect, like coroutine, and on_error maps an exception raised by the thunk or by the await into the typed error channel. Re-raise from on_error to keep an unexpected exception a defect. run_main and the run_async family can interpret the result.

Running effects

E.MissingRequirement

@final
@dataclass(frozen=True)
class MissingRequirement(Exception):
    requirement_type: TypeForm[Any]

Defect for a requirement requested at runtime without being provided.

Unreachable through fully typed code. run_sync_exit settles as Failure(Die(MissingRequirement(...))); run_sync raises it.

E.AsyncEffectInSyncRun

@final
@dataclass(frozen=True)
class AsyncEffectInSyncRun(Exception):
    ...

Defect for a coroutine effect reached by a synchronous runner.

run_main and the run_async family can await. run_sync_exit settles as Failure(Die(AsyncEffectInSyncRun())); run_sync raises it.

E.run_sync

def run_sync[A, E: EffectonError](effect: Effect[A, E]) -> A

Interpret an effect and return its value, raising on failure.

A typed failure raises the error itself, a defect re-raises the exception (or UnhandledDefect for a non-exception value) and an interruption re-raises the exception that signalled it. Use run_sync_exit to receive the Exit instead.

E.run_sync_exit

def run_sync_exit[A, E: EffectonError](effect: Effect[A, E]) -> Exit[A, E]

Interpret an effect and return its Exit.

Coroutine effects are not awaited: reaching one settles the run as Failure(Die(AsyncEffectInSyncRun())), and finalizers still run. The Clock is the blocking SyncLive unless the effect provides another.

E.run_async

def run_async[A, E: EffectonError](effect: Effect[A, E]) -> A

Run an effect on a fresh asyncio loop and return its value.

Owns the event loop through asyncio.run, so it cannot be called from a running loop; use run_async_coroutine there. A typed failure raises the error itself, a defect re-raises the exception (or UnhandledDefect for a non-exception value) and an interruption re-raises the exception that signalled it. Use run_async_exit to receive the Exit instead.

E.run_async_exit

def run_async_exit[A, E: EffectonError](effect: Effect[A, E]) -> Exit[A, E]

Run an effect on a fresh asyncio loop and return its Exit.

Owns the event loop through asyncio.run, so it cannot be called from a running loop; use run_async_coroutine there.

E.run_async_coroutine

def run_async_coroutine[A, E: EffectonError](effect: Effect[A, E]) -> Exit[A, E]

Interpret an effect under asyncio, awaiting every coroutine effect.

This is the coroutine form for a caller that already owns a loop: pass it to asyncio.run, create_task or await it directly.

A cancellation, or any other BaseException raised by an await, a thunk or a callback, unwinds the effect with an Interrupt cause so finalizers run, and the run settles as Failure(Interrupt(exception)). The cancellation is consumed: a caller whose task should stop re-raises the carried exception. Finalizers are shielded: a cancellation that arrives while one is awaiting is remembered, the finalizer runs to completion, and the interruption is applied once it settles. The Clock is the awaiting AsyncLive unless the effect provides another.

E.run_main

def run_main[A, E: EffectonError](effect: Effect[A, E]) -> A

Run a main program, returning its value or reporting and exiting.

Owns a fresh asyncio loop and temporarily handles SIGINT and SIGTERM; call only from the main thread, outside an existing event loop. Signals cooperatively cancel the program and wait for its finalizers, without a cleanup timeout. Blocking synchronous work can delay cancellation.

Unhandled failures use the default effecton logger, independently of program-local logging requirements. Errors and defects exit with their integer exit_code attribute in 0..255, or 1. Interruptions are quiet: SIGTERM exits with 143, other interruptions with 130. An explicit SystemExit retains its code. Successful integers are values, not exit codes.

E.Succeeded

@final
@dataclass(frozen=True)
class Succeeded(Generic[A]):
    value: A
    kind: Literal['succeeded'] = 'succeeded'

E.Failure

@final
@dataclass(frozen=True)
class Failure(Generic[E]):
    cause: Cause[E]
    kind: Literal['failure'] = 'failure'

E.Exit

type Exit[A, E: EffectonError = Never] = Succeeded[A] | Failure[E]

E.UnhandledDefect

@final
@dataclass(frozen=True)
class UnhandledDefect(Exception):
    defect: Any

Raised by a throwing runner for a defect that is not an exception.

Exception defects are re-raised as they are; any other value, such as the argument of die("boom"), is wrapped here so it can still propagate through Python's exception machinery.

Requirements

E.ImplicitRequirement

@runtime_checkable
class ImplicitRequirement(Protocol):
    ...

A requirement that carries its own default value.

The default is computed once per process and shared by every later interpretation, so it must be an immutable value.

E.ImplicitRequirement.default

def default() -> Self

E.require_implicit

def require_implicit[S: ImplicitRequirement](requirement_type: type[S]) -> Effect[S]

E.provide_implicit

def provide_implicit[S: ImplicitRequirement, A, E: EffectonError, R](
    effect: Effect[A, E, R],
    value: S,
) -> Effect[A, E, R]

Scope

E.Scope

@final
@dataclass(frozen=True)
class Scope:
    ...

E.Scope.add_finalizer

def add_finalizer(finalizer: Effect[Any]) -> None

E.Scope.close

def close() -> Effect[None]

E.add_finalizer

def add_finalizer(finalizer: Effect[Any]) -> Effect[None, Never, Scope]

E.scoped

def scoped[A, E: EffectonError, R = Never](
    effect: Effect[A, E, Scope | R],
) -> Effect[A, E, R]

E.acquire_and_release

def acquire_and_release[A, E: EffectonError, R](
    acquire: Effect[A, E, R],
    release: Callable[[A], Effect[Any]],
) -> Effect[A, E, R | Scope]

Acquire a resource whose release the enclosing scope guarantees.

The release is registered only after acquire succeeds; a failed or dying acquire registers nothing. suspend defers the release effect's construction, so a release function that raises does so at close time as a finalizer defect that cannot skip other finalizers.

Fibers

E.fork

def fork[A, E: EffectonError](effect: Effect[A, E]) -> Effect[Fiber[A, E]]

Start effect as a concurrent run and return its Fiber.

The fiber starts on the next turn of the loop; yielding to the loop, for example through a Test clock move, lets it reach its first suspension before the caller continues.

E.yield_now

def yield_now() -> Effect[None]

Let every other runnable fiber take a turn before continuing.

E.Fiber

@final
@dataclass(frozen=True)
class Fiber[A, E: EffectonError]:
    task: asyncio.Task[Exit[A, E]]

E.Fiber.join

def join() -> Effect[A, E]

The fiber's value, failing with the fiber's own cause.

E.Fiber.wait

def wait() -> Effect[Exit[A, E]]

The fiber's Exit once it settles; never fails itself.

E.Fiber.poll

def poll() -> Effect[Exit[A, E] | None]

The fiber's Exit if it has settled, else None, without waiting.

E.Fiber.interrupt

def interrupt() -> Effect[Exit[A, E]]

Cancel the fiber, let its finalizers run and return its Exit.

E.race_first

def race_first[A, E: EffectonError, R, B, E2: EffectonError, R2](
    left: Effect[A, E, R],
    right: Effect[B, E2, R2],
) -> Effect[A | B, E | E2, R | R2]

Return the first completed outcome, interrupting and awaiting the loser.

Both branches inherit surrounding requirements and start when the race runs. Failure, defects, and interruption can win; the left wins when both are complete at selection. Loser cleanup cannot replace the winner. Parent cancellation waits for both branches' finalizers. Async only.

Racing Fiber.join() cancels only the waiter, not the independent fiber.

Timing

E.timeout

def timeout(duration: timedelta | None = None, **parts: Unpack[Parts]) -> Timeout

Bind the deadline, a timedelta or its parts (seconds=...); apply the result to an effect or a function.

E.TimeoutException

@final
@dataclass(frozen=True)
class TimeoutException(EffectonError):
    duration: timedelta

E.Schedule

@final
@dataclass(frozen=True)
class Schedule:
    steps: Callable[[], Iterator[Effect[timedelta]]]

A recurrence policy: how often to recur and how long to wait first.

The constructors cover the basic policies; from_delays builds a custom one from any callable that yields a fresh iterator of delays, and passing steps directly admits delays computed by effects.

E.Schedule.from_delays

def from_delays(delays: Callable[[], Iterator[timedelta]]) -> Schedule

A schedule whose steps are the plain delays the callable yields.

E.Schedule.recurs

def recurs(times: int) -> Schedule

Recur up to times more times, without waiting.

E.Schedule.spaced

def spaced(delay: timedelta | None = None, **parts: Unpack[Parts]) -> Schedule

Recur forever, waiting a timedelta or its parts (seconds=...) each time.

E.Schedule.exponential

def exponential(
    base: timedelta | None = None,
    factor: float = 2.0,
    **parts: Unpack[Parts],
) -> Schedule

Recur forever, waiting base, then base * factor, and so on.

base is a timedelta or its parts: exponential(seconds=0.1).

E.Schedule.jittered

def jittered(*, min: float = 0.8, max: float = 1.2) -> Schedule

Scale each delay by a factor drawn uniformly from [min, max].

The factor comes from the Random service, so providing Random.Test makes the jitter deterministic.

Timeout

@final
@dataclass(frozen=True)
class Timeout:
    duration: timedelta

Not exported as E.Timeout; reached through the return types above.

One step of timeout(duration)(...): the deadline is bound.

Timeout.call

@overload
def __call__[A, E: EffectonError, R](
    effect: Effect[A, E, R],
) -> Effect[A, E | TimeoutException, R]
 
@overload
def __call__[**P, A, E: EffectonError, R](
    f: Callable[P, Effect[A, E, R]],
) -> Callable[P, Effect[A, E | TimeoutException, R]]

Parts

class Parts(TypedDict):
    weeks: float
    days: float
    hours: float
    minutes: float
    seconds: float
    milliseconds: float
    microseconds: float

Not exported as E.Parts; reached through the return types above.

The keyword arguments of timedelta.

Logging

E.LogLevel

@final
class LogLevel(Enum):
    ALL = 'all'
    TRACE = 'trace'
    DEBUG = 'debug'
    INFO = 'info'
    WARN = 'warn'
    ERROR = 'error'
    FATAL = 'fatal'
    NONE = 'none'

E.Severity

type Severity = Literal[LogLevel.TRACE, LogLevel.DEBUG, LogLevel.INFO, LogLevel.WARN, LogLevel.ERROR, LogLevel.FATAL]

Levels a message can be logged at.

Excludes the ALL and NONE sentinels, which are only meaningful as a MinimumLogLevel threshold: ALL passes everything, NONE silences everything.

E.LogData

@final
@dataclass(frozen=True)
class LogData:
    message: tuple[object, ...]
    log_level: Severity
    date: datetime
    annotations: Mapping[str, object]

E.EffectonLogger

@final
@dataclass(frozen=True)
class EffectonLogger:
    log: Callable[[LogData], None]

E.CurrentLoggers

@final
@dataclass(frozen=True)
class CurrentLoggers(ImplicitRequirement):
    loggers: tuple[EffectonLogger, ...]

E.CurrentLoggers.default

def default() -> CurrentLoggers

E.MinimumLogLevel

@final
@dataclass(frozen=True)
class MinimumLogLevel(ImplicitRequirement):
    level: LogLevel

E.MinimumLogLevel.default

def default() -> MinimumLogLevel

E.CurrentLogLevel

@final
@dataclass(frozen=True)
class CurrentLogLevel(ImplicitRequirement):
    level: Severity

The level a bare log(...) call logs at.

E.CurrentLogLevel.default

def default() -> CurrentLogLevel

E.CurrentLogAnnotations

@final
@dataclass(frozen=True)
class CurrentLogAnnotations(ImplicitRequirement):
    annotations: Mapping[str, object]

E.CurrentLogAnnotations.default

def default() -> CurrentLogAnnotations

E.log

def log(*message: object) -> Effect[None]

E.log_trace

def log_trace(*message: object) -> Effect[None]

E.log_debug

def log_debug(*message: object) -> Effect[None]

E.log_info

def log_info(*message: object) -> Effect[None]

E.log_warning

def log_warning(*message: object) -> Effect[None]

E.log_error

def log_error(*message: object) -> Effect[None]

E.log_fatal

def log_fatal(*message: object) -> Effect[None]

E.annotate_logs

def annotate_logs[A, E: EffectonError, R](
    effect: Effect[A, E, R],
    **annotations: object,
) -> Effect[A, E, R]

Merge annotations into every log call inside the wrapped effect.

E.PrettyFormatter

@final
class PrettyFormatter(logging.Formatter):
    ...

Formats records as [HH:MM:SS.mmm] LEVEL message.

Levels render in effecton's severity vocabulary (WARN, FATAL) with a color per level; colors=None detects whether stderr is a terminal. effecton annotations travel on the record as the effecton_annotations attribute (through extra) and each renders as one indented key: value line. The Clock time travels as effecton_date and, when present, replaces the record's own timestamp in the stamp; plain stdlib records fall back to record.created.

E.PrettyFormatter.format

def format(record: logging.LogRecord) -> str

E.pretty_logger

pretty_logger: EffectonLogger

Path

E.Path

@final
@dataclass(frozen=True, init=False, repr=False)
class Path:
    ...

E.Path.parent

parent: Path

The directory holding this path; the root is its own parent.

E.Path.parents

parents: tuple[Path, ...]

Every ancestor, nearest first; empty for the root.

E.Path.name

name: str

E.Path.suffix

suffix: str

E.Path.stem

stem: str

E.Path.parts

parts: tuple[str, ...]

Schema

E.Schema

Schema: describe a shape once, then decode and encode through it.

A Schema[A, I] pairs two pure functions: decode turns untrusted input into a typed A, encode turns an A back into its wire form I. Combinators build bigger pairs from smaller ones, so every schema works in both directions. decode and encode return effects that fail with one ParseError listing every issue found, each tagged with the path where it occurred.

E.Schema.IssuePath

type IssuePath = tuple[str | int, ...]

E.Schema.TypeMismatch

@final
@dataclass(frozen=True)
class TypeMismatch:
    path: IssuePath
    expected: str
    actual: object

E.Schema.MissingKey

@final
@dataclass(frozen=True)
class MissingKey:
    path: IssuePath

E.Schema.RefinementFailed

@final
@dataclass(frozen=True)
class RefinementFailed:
    path: IssuePath
    message: str
    actual: object

E.Schema.TransformFailed

@final
@dataclass(frozen=True)
class TransformFailed:
    path: IssuePath
    message: str
    actual: object

E.Schema.NoUnionMember

@final
@dataclass(frozen=True)
class NoUnionMember:
    path: IssuePath
    actual: object
    issues: tuple[Issue, ...]

No member matched; issues holds what every member reported.

E.Schema.InvalidJson

@final
@dataclass(frozen=True)
class InvalidJson:
    path: IssuePath
    reason: str

E.Schema.Issue

type Issue = TypeMismatch | MissingKey | RefinementFailed | TransformFailed | NoUnionMember | InvalidJson

E.Schema.ParseError

@final
@dataclass(frozen=True)
class ParseError(EffectonError):
    issues: tuple[Issue, ...]

E.Schema.Schema

@final
class Schema[A, I]:
    ...

A two-way codec: A is the decoded type, I the encoded one.

E.Schema.Schema.check

def check(*checks: Check[A]) -> Schema[A, I]

Guard decode and encode alike: every check must hold in both directions.

Encode runs the inner schema first, so a value of a foreign type is a TypeMismatch rather than a crash inside a predicate.

E.Schema.Check

@final
class Check(Generic[_T_contra]):
    ...

A refinement: a predicate over the decoded value and its failure message.

E.Schema.decode

@overload
def decode[T: Struct](schema: type[T]) -> Callable[[object], Effect[T, ParseError]]
 
@overload
def decode[A, I](schema: Schema[A, I]) -> Callable[[object], Effect[A, ParseError]]

decode(schema)(raw): turn untrusted input into a typed value.

E.Schema.encode

@overload
def encode[T: Struct](
    schema: type[T],
) -> Callable[[T], Effect[dict[str, object], ParseError]]
 
@overload
def encode[A, I](schema: Schema[A, I]) -> Callable[[A], Effect[I, ParseError]]

encode(schema)(value): turn a typed value back into its wire form.

E.Schema.decode_json

@overload
def decode_json[T: Struct](schema: type[T]) -> Callable[[str], Effect[T, ParseError]]
 
@overload
def decode_json[A, I](schema: Schema[A, I]) -> Callable[[str], Effect[A, ParseError]]

decode_json(schema)(text): parse JSON text, then decode it.

E.Schema.encode_json

@overload
def encode_json[T: Struct](schema: type[T]) -> Callable[[T], Effect[str, ParseError]]
 
@overload
def encode_json[A, I](schema: Schema[A, I]) -> Callable[[A], Effect[str, ParseError]]

encode_json(schema)(value): encode, then serialize as JSON text.

An encoded form json cannot serialize is a defect: pick schemas whose wire side is JSON (every built-in is).

E.Schema.field

@overload
def field[T](schema: Schema[T, Any], *, key: str | None = None) -> T
 
@overload
def field[T](schema: Schema[T, Any], *, key: str | None = None, default: T) -> T
 
@overload
def field(*, key: str | None = None) -> Any
 
@overload
def field[T](*, key: str | None = None, default: T) -> T

Configure a Struct field: its schema, its wire key, its default.

A field with a default may be absent from the input; the default is checked against the schema once, when the class is defined. Without a schema, one is inferred from the annotation.

E.Schema.Struct

@dataclass_transform(frozen_default=True, kw_only_default=True, field_specifiers=(field,))
class Struct:
    ...

Subclass to declare a record: annotations are the decoded types.

Every subclass becomes a frozen, keyword-only dataclass and is itself accepted wherever a schema is: decode(User)(raw).

E.Schema.struct_schema

def struct_schema[T: Struct](cls: type[T]) -> Schema[T, dict[str, object]]

The schema behind a Struct class, for combinators that need one.

E.Schema.Literal

def Literal[T: str | int | bool | None](*values: T) -> Schema[T, T]

One of the given values; True is not 1.

E.Schema.instance_of

def instance_of[T](cls: type[T]) -> Schema[T, T]

Any instance of cls, unchanged in both directions.

The guard for a transform's decoded side: pass it as to= so encode rejects a foreign value instead of handing it to your function.

E.Schema.Array

@overload
def Array[T: Struct](item: type[T]) -> Schema[tuple[T, ...], list[dict[str, object]]]
 
@overload
def Array[A, I](item: Schema[A, I]) -> Schema[tuple[A, ...], list[I]]

A list or tuple on the wire, a tuple once decoded.

E.Schema.Record

def Record[KA, KI, VA, VI](
    key: Schema[KA, KI],
    value: Schema[VA, VI],
) -> Schema[dict[KA, VA], dict[KI, VI]]

A mapping whose keys and values each go through their schema.

E.Schema.Tuple

@overload
def Tuple[A1, I1](m1: Schema[A1, I1], /) -> Schema[tuple[A1], list[I1]]
 
@overload
def Tuple[A1, I1, A2, I2](
    m1: Schema[A1, I1],
    m2: Schema[A2, I2],
    /,
) -> Schema[tuple[A1, A2], list[I1 | I2]]
 
@overload
def Tuple[A1, I1, A2, I2, A3, I3](
    m1: Schema[A1, I1],
    m2: Schema[A2, I2],
    m3: Schema[A3, I3],
    /,
) -> Schema[tuple[A1, A2, A3], list[I1 | I2 | I3]]
 
@overload
def Tuple[A1, I1, A2, I2, A3, I3, A4, I4](
    m1: Schema[A1, I1],
    m2: Schema[A2, I2],
    m3: Schema[A3, I3],
    m4: Schema[A4, I4],
    /,
) -> Schema[tuple[A1, A2, A3, A4], list[I1 | I2 | I3 | I4]]
 
@overload
def Tuple(*members: Schema[Any, Any]) -> Schema[tuple[Any, ...], list[Any]]

A fixed-length array with one schema per position.

E.Schema.Union

@overload
def Union[A1, I1, A2, I2](
    m1: Schema[A1, I1],
    m2: Schema[A2, I2],
    /,
) -> Schema[A1 | A2, I1 | I2]
 
@overload
def Union[A1, I1, A2, I2, A3, I3](
    m1: Schema[A1, I1],
    m2: Schema[A2, I2],
    m3: Schema[A3, I3],
    /,
) -> Schema[A1 | A2 | A3, I1 | I2 | I3]
 
@overload
def Union[A1, I1, A2, I2, A3, I3, A4, I4](
    m1: Schema[A1, I1],
    m2: Schema[A2, I2],
    m3: Schema[A3, I3],
    m4: Schema[A4, I4],
    /,
) -> Schema[A1 | A2 | A3 | A4, I1 | I2 | I3 | I4]
 
@overload
def Union(*members: Schema[Any, Any]) -> Schema[Any, Any]

The first member that succeeds wins, decoding and encoding alike.

E.Schema.NullOr

@overload
def NullOr[T: Struct](schema: type[T]) -> Schema[T | None, dict[str, object] | None]
 
@overload
def NullOr[A, I](schema: Schema[A, I]) -> Schema[A | None, I | None]

The schema or null, in both directions.

E.Schema.Invalid

@final
@dataclass(frozen=True)
class Invalid:
    message: str

What a transform_or_fail function returns to reject its input.

E.Schema.transform

def transform[A, I, B](
    from_: Schema[A, I],
    *,
    decode: Callable[[A], B],
    encode: Callable[[B], A],
    to: Schema[B, B] | None = None,
) -> Schema[B, I]

Map a schema's decoded side through a total function pair.

Without to, encode hands the value straight to the function, so inside a Union or NullOr the function can receive values meant for another member; pass to to guard it, typically S.instance_of(cls).

E.Schema.transform_or_fail

def transform_or_fail[A, I, B](
    from_: Schema[A, I],
    *,
    decode: Callable[[A], B | Invalid],
    encode: Callable[[B], A | Invalid],
    to: Schema[B, B] | None = None,
) -> Schema[B, I]

Like transform, but either function may return Invalid(message).

An exception raised by either function is a defect, not a ParseError. to guards the decoded side in both directions, as in transform.

E.Schema.filter

def filter[A](predicate: Callable[[A], bool], *, message: str) -> Check[A]

A refinement for Schema.check(): a bare predicate plus its failure message.

E.Schema.pattern

def pattern(regex: str) -> Check[str]

A string the regular expression finds a match in.

E.Schema.String

String: Schema[str, str] = _primitive('string', lambda x: isinstance(x, str))

E.Schema.Int

Int: Schema[int, int] = _primitive('integer', lambda x: isinstance(x, int) and not isinstance(x, bool))

E.Schema.Float

Float: Schema[float, float] = _primitive('number', lambda x: isinstance(x, int | float) and not isinstance(x, bool))

E.Schema.Bool

Bool: Schema[bool, bool] = _primitive('boolean', lambda x: isinstance(x, bool))

E.Schema.Null

Null: Schema[None, None] = _primitive('null', lambda x: x is None)

E.Schema.Unknown

Unknown: Schema[object, object] = _primitive('anything', lambda _: True)

E.Schema.IntFromString

IntFromString: Schema[int, str] = transform_or_fail(String, decode=_parser(int, 'expected an integer string'), encode=str, to=Int)

E.Schema.FloatFromString

FloatFromString: Schema[float, str] = transform_or_fail(String, decode=_parser(float, 'expected a number string'), encode=str, to=Float)

E.Schema.DateTimeFromString

DateTimeFromString: Schema[datetime, str] = transform_or_fail(String, decode=_parser(datetime.fromisoformat, 'expected an ISO 8601 datetime'), encode=datetime.isoformat, to=instance_of(datetime))

E.Schema.DateFromString

DateFromString: Schema[date, str] = transform_or_fail(String, decode=_parser(date.fromisoformat, 'expected an ISO 8601 date'), encode=date.isoformat, to=instance_of(date).check(filter(lambda d: not isinstance(d, datetime), message='expected a date without a time')))

E.Schema.PathFromString

PathFromString: Schema[Path, str] = transform(String, decode=Path, encode=str, to=instance_of(Path))

Cli

E.Cli

Cli: declare a command's arguments as an annotated class, then parse argv through Schema and run the handler as an effect.

An Args subclass is a frozen keyword-only dataclass: annotations are the decoded types and Annotated metadata (Option, Argument) describes the command line. Each field gets a text codec, a Schema whose wire side is str, so the arguments decode like any other struct and every problem is reported together as one usage error. Commands nest with with_subcommands, and run(command) is one Effect that reads argv through E.Process, prints help and the version, and fails with a UsageError that exits with status 2 under run_main.

E.Cli.run

def run[E: EffectonError, R](
    command: Command[E, R],
) -> Effect[None, E | UsageError, R | process.Protocol]

Read argv through E.Process, parse it against the command tree, run the handler.

--help prints the command's help and --version, on the root, its distribution's version; both succeed. A parsing problem fails with a UsageError, which exits with status 2 under run_main.

E.Cli.command

@overload
def command[T: Args, E: EffectonError, R](
    name: str,
    *,
    help: str = '',
    args: type[T],
    handler: Callable[[T], Effect[None, E, R]],
) -> Command[E, R]
 
@overload
def command[E: EffectonError, R](
    name: str,
    *,
    help: str = '',
    handler: Callable[[], Effect[None, E, R]],
) -> Command[E, R]
 
@overload
def command(name: str, *, help: str = '') -> Command[Never, Never]

A command: a handler over an Args class, a handler alone, or a parent for with_subcommands. The caller's module is recorded for --version.

E.Cli.Command

@final
@dataclass(frozen=True)
class Command(Generic[_E, _R]):
    name: str
    help: str

A named command: E and R are the union over every handler beneath it.

E.Cli.Command.with_subcommands

def with_subcommands[E2: EffectonError, R2](
    *commands: Command[E2, R2],
) -> Command[_E | E2, _R | R2]

A copy of this command that dispatches to the given subcommands.

E.Cli.Option

@final
@dataclass(frozen=True)
class Option:
    help: str = ''
    name: str | None = None
    short: str | None = None
    metavar: str | None = None
    schema: S.Schema[Any, str] | None = None

Annotated metadata for an option field: --name VALUE, or a flag for bool.

E.Cli.Argument

@final
@dataclass(frozen=True)
class Argument:
    help: str = ''
    name: str | None = None
    metavar: str | None = None
    schema: S.Schema[Any, str] | None = None

Annotated metadata for a positional argument field.

E.Cli.Args

@dataclass_transform(frozen_default=True, kw_only_default=True)
class Args:
    ...

Subclass to declare a command's arguments: annotations are the decoded types.

Every subclass becomes a frozen, keyword-only dataclass. A field is an option unless its Annotated metadata is Argument; both carry help text, names and an optional explicit text codec (a Schema whose wire side is str). str, int, float, E.Path, datetime, date and Literal[str...] infer theirs; bool is a flag; T | None is optional; tuple[T, ...] repeats.

E.Cli.UsageError

@final
@dataclass(frozen=True)
class UsageError(EffectonError):
    command: str
    usage: str
    reason: UsageReason
    exit_code: int = 2

The command line could not be parsed; reason says why.

Exits with 2 under run_main.

E.Cli.UsageReason

type UsageReason = UnknownOption | UnknownCommand | MissingCommand | MissingOptionValue | UnexpectedOptionValue | UnexpectedArgument | InvalidArguments

E.Cli.UnknownOption

@final
@dataclass(frozen=True)
class UnknownOption:
    option: str

E.Cli.UnknownCommand

@final
@dataclass(frozen=True)
class UnknownCommand:
    name: str

E.Cli.MissingCommand

@final
@dataclass(frozen=True)
class MissingCommand:
    ...

E.Cli.MissingOptionValue

@final
@dataclass(frozen=True)
class MissingOptionValue:
    option: str

E.Cli.UnexpectedOptionValue

@final
@dataclass(frozen=True)
class UnexpectedOptionValue:
    option: str
    value: str

E.Cli.UnexpectedArgument

@final
@dataclass(frozen=True)
class UnexpectedArgument:
    argument: str

E.Cli.InvalidArguments

@final
@dataclass(frozen=True)
class InvalidArguments:
    issues: tuple[S.Issue, ...]

Clock

E.Clock

Clock service: Protocol plus two live clocks and a Test clock.

The Protocol is an implicit requirement: programs read the time through E.now() and pause through E.sleep() without declaring anything in R. Sleeping depends on the runner, so there are two live clocks and each runner installs the matching one: run_sync installs SyncLive, whose sleep blocks the thread, and the run_async family installs AsyncLive, whose sleep awaits asyncio.sleep so the loop keeps turning and a cancellation interrupts it. Tests override either with .provide(Protocol)(Test(...)), whose sleep is async only and parks until adjust or set_time moves the clock to or past the wake time.

The accessors live here as _now and _sleep and are exported only as E.now and E.sleep, so there is one way to reach the clock rather than both E.now() and E.Clock.now().

E.now

def now() -> Effect[datetime]

The current time as a timezone-aware UTC datetime, read from the Clock.

E.sleep

def sleep(duration: timedelta | None = None, **parts: Unpack[Parts]) -> Effect[None]

Pause through the Clock for a timedelta or its parts (seconds=...).

How the pause happens depends on the runner.

E.Clock.Protocol

@runtime_checkable
class Protocol(ImplicitRequirement, typing.Protocol):
    ...

E.Clock.Protocol.now

def now() -> Effect[datetime]

E.Clock.Protocol.sleep

def sleep(duration: timedelta) -> Effect[None]

E.Clock.Protocol.default

def default() -> Never

E.Clock.SyncLive

@final
@dataclass(frozen=True)
class SyncLive(Protocol):
    ...

The wall clock with a blocking sleep; run_sync installs it.

E.Clock.AsyncLive

@final
@dataclass(frozen=True)
class AsyncLive(Protocol):
    ...

The wall clock with an awaiting sleep; the run_async family installs it.

E.Clock.Test

@final
@dataclass
class Test(Protocol):
    current: datetime = datetime(1970, 1, 1, tzinfo=UTC)

A clock that only moves when told to.

Pass current to start elsewhere; the default is the Unix epoch. sleep is async only: it parks until adjust or set_time moves the clock to or past the wake time. Both movers are effects, so a test written as an effect moves the clock with yield from, and the effecton pytest plugin provides this clock to such a test through its test_clock fixture.

E.Clock.Test.adjust

def adjust(delta: timedelta) -> Effect[None]

E.Clock.Test.set_time

def set_time(time: datetime) -> Effect[None]

Move the clock, waking every sleeper whose wake time has come.

Async only, like sleep. The move yields to the loop before and after, so a program forked just before reaches its sleep and registers, and the woken programs progress before the caller continues.

Random

E.Random

Random service: Protocol plus a Live generator and a seeded Test one.

The Protocol is an implicit requirement: programs draw randomness by resolving the service with E.random() and calling its methods, which keep the names of the random standard library, without declaring anything in R. Live is the default and delegates to the module-level functions, so it shares the process-global generator; it holds no state of its own, which is what lets the interpreter memoize it once per process. Tests override it with .provide(Protocol)(Test(seed)), whose private generator makes every draw a function of the seed.

The accessor lives here as _random and is exported only as E.random, so there is one way to reach the service rather than both E.random() and E.Random.random().

E.random

def random() -> Effect[Protocol]

The Random service, resolved from the environment or its default.

E.Random.Protocol

@runtime_checkable
class Protocol(ImplicitRequirement, typing.Protocol):
    ...

E.Random.Protocol.random

def random() -> Effect[float]

A float in [0.0, 1.0).

E.Random.Protocol.uniform

def uniform(a: float, b: float) -> Effect[float]

A float in [a, b].

E.Random.Protocol.randint

def randint(a: int, b: int) -> Effect[int]

An int in [a, b], both ends included.

E.Random.Protocol.choice

def choice[T](seq: Sequence[T]) -> Effect[T]

One element of a non-empty sequence.

E.Random.Protocol.shuffle

def shuffle[T](seq: Sequence[T]) -> Effect[list[T]]

A new list with the elements of seq in random order; seq is untouched.

E.Random.Protocol.default

def default() -> Protocol

E.Random.Live

@final
@dataclass(frozen=True)
class Live(Protocol):
    ...

The process-global generator behind the random module's functions.

E.Random.Test

@final
@dataclass
class Test(Protocol):
    seed: int = 0

A generator seeded once, so the same seed always yields the same draws.

The default seed is 0. Each instance advances on its own: provide one instance to every effect whose draws should form a single sequence.

FileSystem

E.FileSystem

FileSystem service: Protocol plus SyncLive, AsyncLive and an in-memory Test.

Paths are E.Path values, and this module is the only place where they touch the disk. SyncLive calls the os-level standard library (os.*, open) and blocks the thread, so it suits run_sync; AsyncLive makes the same calls through aiofiles, so the loop keeps turning under run_async and run_main. Both share one error mapping: the failures a program reacts to (a missing file, permissions, a directory where a file should be and the reverse, a path already taken, a directory that is not empty) are typed, and everything else, such as disk full or an I/O error, stays a defect. Test keeps the tree in dicts and enforces the same rules, so a program sees the same Exit whichever implementation it runs against.

Naming follows Effect-TS's FileSystem with these deliberate differences: read_directory returns full paths rather than names, exists does not follow symlinks (a dangling link exists), symlink takes (target, link) like os.symlink; the working and home directories are read through the Process service, since a Path has no way to reach the environment.

E.FileSystem.FileNotFound

@final
@dataclass(frozen=True)
class FileNotFound(EffectonError):
    path: Path

E.FileSystem.PermissionDenied

@final
@dataclass(frozen=True)
class PermissionDenied(EffectonError):
    path: Path

E.FileSystem.PathIsADirectory

@final
@dataclass(frozen=True)
class PathIsADirectory(EffectonError):
    path: Path

E.FileSystem.PathIsNotADirectory

@final
@dataclass(frozen=True)
class PathIsNotADirectory(EffectonError):
    path: Path

E.FileSystem.PathAlreadyExists

@final
@dataclass(frozen=True)
class PathAlreadyExists(EffectonError):
    path: Path

E.FileSystem.DirectoryNotEmpty

@final
@dataclass(frozen=True)
class DirectoryNotEmpty(EffectonError):
    path: Path

E.FileSystem.FileSystemError

type FileSystemError = FileNotFound | PermissionDenied | PathIsADirectory | PathIsNotADirectory | PathAlreadyExists | DirectoryNotEmpty

E.FileSystem.StatError

type StatError = FileNotFound | PermissionDenied

E.FileSystem.ReadError

type ReadError = FileNotFound | PermissionDenied | PathIsADirectory

E.FileSystem.WriteError

type WriteError = FileNotFound | PermissionDenied | PathIsADirectory | PathIsNotADirectory

E.FileSystem.MakeDirectoryError

type MakeDirectoryError = FileNotFound | PermissionDenied | PathAlreadyExists | PathIsNotADirectory

E.FileSystem.ReadDirectoryError

type ReadDirectoryError = FileNotFound | PermissionDenied | PathIsNotADirectory

E.FileSystem.RemoveError

type RemoveError = FileNotFound | PermissionDenied | DirectoryNotEmpty

E.FileSystem.RenameError

type RenameError = FileNotFound | PermissionDenied | PathIsADirectory | PathIsNotADirectory | DirectoryNotEmpty

E.FileSystem.CopyFileError

type CopyFileError = FileNotFound | PermissionDenied | PathIsADirectory

E.FileSystem.SymlinkError

type SymlinkError = FileNotFound | PermissionDenied | PathAlreadyExists | PathIsNotADirectory

E.FileSystem.FileType

type FileType = Literal['file', 'directory', 'symlink', 'other']

E.FileSystem.FileInfo

@final
@dataclass(frozen=True)
class FileInfo:
    type: FileType
    size: int
    modified_at: datetime

What lstat reports about a path: the entry itself, not a link's target.

E.FileSystem.Protocol

@runtime_checkable
class Protocol(typing.Protocol):
    ...

E.FileSystem.Protocol.exists

def exists(path: Path) -> Effect[bool, PermissionDenied]

Whether the entry exists; a dangling symlink counts.

E.FileSystem.Protocol.stat

def stat(path: Path) -> Effect[FileInfo, StatError]

E.FileSystem.Protocol.read_file

def read_file(path: Path) -> Effect[bytes, ReadError]

E.FileSystem.Protocol.read_file_string

def read_file_string(path: Path, encoding: str = 'utf-8') -> Effect[str, ReadError]

The file decoded; undecodable bytes are a defect.

E.FileSystem.Protocol.write_file

def write_file(path: Path, content: bytes) -> Effect[None, WriteError]

Create or overwrite the file; the parent directory must exist.

E.FileSystem.Protocol.write_file_string

def write_file_string(
    path: Path,
    content: str,
    encoding: str = 'utf-8',
) -> Effect[None, WriteError]

E.FileSystem.Protocol.make_directory

def make_directory(
    path: Path,
    *,
    recursive: bool = False,
) -> Effect[None, MakeDirectoryError]

Create the directory; recursive also creates parents and accepts an existing directory, while a file at the path is PathAlreadyExists either way.

E.FileSystem.Protocol.read_directory

def read_directory(path: Path) -> Effect[tuple[Path, ...], ReadDirectoryError]

The entries as full paths, sorted.

E.FileSystem.Protocol.remove

def remove(path: Path, *, recursive: bool = False) -> Effect[None, RemoveError]

Remove a file, a symlink (never its target) or a directory, which must be empty unless recursive.

E.FileSystem.Protocol.rename

def rename(old: Path, new: Path) -> Effect[None, RenameError]

Move an entry, replacing a file or an empty directory at new.

E.FileSystem.Protocol.copy_file

def copy_file(src: Path, dst: Path) -> Effect[None, CopyFileError]

E.FileSystem.Protocol.symlink

def symlink(target: Path, link: Path) -> Effect[None, SymlinkError]

Create link pointing at target, which need not exist.

E.FileSystem.Protocol.read_link

def read_link(link: Path) -> Effect[Path, StatError]

The target a symlink points at; a non-link is a defect.

E.FileSystem.SyncLive

@final
@dataclass(frozen=True)
class SyncLive(Protocol):
    ...

The real file system through blocking os calls; suits run_sync.

E.FileSystem.AsyncLive

@final
@dataclass(frozen=True)
class AsyncLive(Protocol):
    ...

The real file system through aiofiles; suits run_async and run_main.

Every call awaits aiofiles' twin of the os call SyncLive makes, so the loop keeps turning while a worker thread does the I/O. Under run_sync the effects die with AsyncEffectInSyncRun, like any coroutine effect; a cancellation abandons the blocking call in its thread rather than aborting it.

E.FileSystem.Test

@final
@dataclass
class Test(Protocol):
    files: dict[Path, bytes | str] = field(default_factory=dict)
    directories: set[Path] = field(default_factory=set)
    links: dict[Path, Path] = field(default_factory=dict)

An in-memory tree that follows the real rules.

Seed it with files (text or bytes), directories and links; every ancestor of a seeded path is created too, and the root always exists. Operations then enforce what the disk would: a parent must exist and be a directory, a directory must be empty to go, and so on, so a program gets the same Exit here as against a Live. Content operations follow a symlink at the final component (a dangling one is FileNotFound); exists, stat, remove, rename and read_link act on the link itself, and links in the middle of a path are not resolved. modified_at is always the epoch.

HttpClient

E.HttpClient

HttpClient service: Protocol plus SyncLive, AsyncLive and a canned Test.

Requests and responses are plain values (Request, Response), and this module is the only place that talks to the network. SyncLive sends each request through a blocking httpx2.Client, so it suits run_sync; AsyncLive sends it through httpx2.AsyncClient, so the loop keeps turning under run_async and run_main. Both share one error mapping: the transport failing (a connection that could not be made or was lost, a proxy refusing, a reply that was not HTTP) is the one typed error, TransportError, as in Effect-TS; everything else stays a defect, including a URL without a scheme, which is a programming error. Test answers from a canned url -> body mapping or from a handler and records every request, so a program sees the same Exit whichever implementation it runs against.

Status semantics follow Effect-TS's HttpClient: every status comes back as a Response, and filter_status_ok is the opt-in that turns a non-2xx into a StatusError. Redirects are not followed and the clients set no HTTP-level timeout, so a slow server is bounded the effecton way, by composing effect.timeout(...). Naming follows Effect-TS with these deliberate differences: delete is spelled out (del is a Python keyword), and Response.json() is an effect that fails with InvalidJson.

E.HttpClient.Method

type Method = Literal['GET', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS']

E.HttpClient.TransportError

@final
@dataclass(frozen=True)
class TransportError(EffectonError):
    method: Method
    url: str
    message: str

The request never got a valid response: the connection could not be made or was lost, a proxy refused, or the reply was not HTTP.

E.HttpClient.StatusError

@final
@dataclass(frozen=True)
class StatusError(EffectonError):
    method: Method
    url: str
    status: int

E.HttpClient.InvalidJson

@final
@dataclass(frozen=True)
class InvalidJson(EffectonError):
    method: Method
    url: str
    message: str

E.HttpClient.HttpClientError

type HttpClientError = TransportError | StatusError | InvalidJson

E.HttpClient.Request

@final
@dataclass(frozen=True)
class Request:
    method: Method
    url: str
    headers: Mapping[str, str] = field(default_factory=dict)
    body: bytes | None = None

What gets sent: the conveniences on Protocol build one, execute takes it.

E.HttpClient.Response

@final
@dataclass(frozen=True)
class Response:
    request: Request
    status: int
    headers: Mapping[str, str] = field(default_factory=dict)
    body: bytes = b''

A fully read response. Header names are lowercase and a header that was sent more than once is comma-joined, as httpx reports them.

E.HttpClient.Response.is_success

is_success: bool

E.HttpClient.Response.text

text: str

The body decoded with the content-type charset, utf-8 when there is none or it is unknown; undecodable bytes become U+FFFD, as in httpx.

E.HttpClient.Response.json

def json() -> Effect[Any, InvalidJson]

E.HttpClient.Protocol

@runtime_checkable
class Protocol(typing.Protocol):
    ...

E.HttpClient.Protocol.execute

def execute(request: Request) -> Effect[Response, TransportError]

Send the request and read the whole response, whatever its status.

E.HttpClient.Protocol.get

def get(
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
) -> Effect[Response, TransportError]

params are url-encoded onto the query string.

E.HttpClient.Protocol.head

def head(
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
) -> Effect[Response, TransportError]

E.HttpClient.Protocol.options

def options(
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
) -> Effect[Response, TransportError]

E.HttpClient.Protocol.delete

def delete(
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
) -> Effect[Response, TransportError]

E.HttpClient.Protocol.post

def post(
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    body: bytes | str | None = None,
    json: object = None,
) -> Effect[Response, TransportError]

A str body is utf-8 encoded; json is serialized and sets content-type: application/json unless a content-type header is given. Giving both body and json is a ValueError.

E.HttpClient.Protocol.put

def put(
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    body: bytes | str | None = None,
    json: object = None,
) -> Effect[Response, TransportError]

E.HttpClient.Protocol.patch

def patch(
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    body: bytes | str | None = None,
    json: object = None,
) -> Effect[Response, TransportError]

E.HttpClient.filter_status_ok

def filter_status_ok(response: Response) -> Effect[Response, StatusError]

The response when its status is 2xx, otherwise a StatusError; Effect-TS's filterStatusOk applied per response: http.get(url).flat_map(filter_status_ok).

E.HttpClient.SyncLive

@final
@dataclass(frozen=True)
class SyncLive(Protocol):
    ...

The real network through a blocking httpx2.Client; suits run_sync.

A fresh client is opened per request (no connection pooling yet) with no HTTP-level timeout, so a request is bounded by composing effect.timeout(...), which needs an async runner.

E.HttpClient.AsyncLive

@final
@dataclass(frozen=True)
class AsyncLive(Protocol):
    ...

The real network through httpx2.AsyncClient; suits run_async and run_main.

Each request opens a fresh client with no HTTP-level timeout, so effect.timeout(...) is how a request gets bounded, and a cancellation aborts the request. Under run_sync the effects die with AsyncEffectInSyncRun, like any coroutine effect.

E.HttpClient.Handler

type Handler = Callable[[Request], Response | TransportError]

E.HttpClient.Test

@final
@dataclass
class Test(Protocol):
    responses: Mapping[str, str | bytes] = field(default_factory=dict)
    handler: Handler | None = None
    requests: list[Request] = field(default_factory=list)

A client that never touches the network.

Seed it with responses, a url -> body mapping answered with a 200 (a str body is utf-8 encoded) where any other url gets an empty 404; or give a handler that receives the normalized Request and returns the Response, or a TransportError to fail with. Either way every executed request is appended to requests, in order, so a test can assert on what was sent.

Process

E.Process

Process service: what the running process knows about its environment.

Today that is the current working directory, the home directory and the command-line arguments; environment variables belong here too when they arrive. Reading them is ambient process state rather than file I/O, so they live apart from FileSystem and E.Path stays a pure value. Like FileSystem it is an explicit requirement: provide Live at the edge, and Test in tests, where all three are plain fields.

E.Process.Protocol

@runtime_checkable
class Protocol(typing.Protocol):
    ...

E.Process.Protocol.cwd

def cwd() -> Effect[Path]

The current working directory.

E.Process.Protocol.home

def home() -> Effect[Path]

The current user's home directory.

E.Process.Protocol.argv

def argv() -> Effect[tuple[str, ...]]

The command-line arguments, without the program name.

E.Process.Live

@final
@dataclass(frozen=True)
class Live(Protocol):
    ...

The real process; its reads are plain sync effects under either runner.

E.Process.Test

@final
@dataclass
class Test(Protocol):
    current_directory: Path = _ROOT
    home_directory: Path = _HOME
    arguments: tuple[str, ...] = ()

Tracer

E.Tracer

Tracer service: Protocol plus a Live tracer and a recording Test one.

A span marks one named, timed unit of a program. with_span opens a span around an effect, makes it the parent of every span opened inside, and ends it with the effect's Exit once the effect settles, on success, failure, defect and interruption alike. Timestamps come from the Clock and ids from the Random service, so the Test clock pins durations and the test_random fixture makes ids reproducible.

The Protocol is an implicit requirement: programs open spans without declaring anything in R. Live is the default; its spans live only as long as the program holds them, so tracing costs nothing until a tracer that exports them is provided. Tests override it with .provide(Protocol)(Test()), which records every span it opens in spans. An OpenTelemetry tracer would be another implementation of the Protocol whose span() returns spans backed by the SDK; the ids are already the hex forms the W3C traceparent header carries.

The accessors live here as _with_span, _annotate_current_span and _current_span and are exported only as E.with_span, E.annotate_current_span and E.current_span, like E.now for the Clock.

E.Tracer.SpanKind

type SpanKind = Literal['internal', 'server', 'client', 'producer', 'consumer']

E.with_span

def with_span[A, E: EffectonError, R](
    effect: Effect[A, E, R],
    name: str,
    *,
    kind: SpanKind = 'internal',
    **attributes: object,
) -> Effect[A, E, R]

Open a span around effect and end it with the effect's Exit.

The span is a child of the current span, if any, and is the current span inside effect. It ends on success, failure, defect and interruption alike.

E.annotate_current_span

def annotate_current_span(**attributes: object) -> EffectGen[None]

Add attributes to the current span; a no-op outside any span.

E.current_span

def current_span() -> Effect[Span, NoCurrentSpan]

The innermost open span, failing with NoCurrentSpan outside any.

E.Tracer.NoCurrentSpan

@final
@dataclass(frozen=True)
class NoCurrentSpan(EffectonError):
    ...

E.Tracer.Started

@final
@dataclass(frozen=True)
class Started:
    start_time: datetime