API Reference
Effect
E.EffectonError
@dataclass(frozen=True)
class EffectonError(Exception):
...E.Fail
@final
@dataclass(frozen=True)
class Fail[E: EffectonError]:
error: EE.Die
@final
@dataclass(frozen=True)
class Die:
defect: AnyE.Interrupt
@final
@dataclass(frozen=True)
class Interrupt:
exception: BaseExceptionThe 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 | InterruptE.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]) -> AInterpret 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]) -> ARun 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]) -> ARun 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: AnyRaised 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() -> SelfE.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]) -> NoneE.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]) -> TimeoutBind 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: timedeltaE.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]]) -> ScheduleA schedule whose steps are the plain delays the callable yields.
E.Schedule.recurs
def recurs(times: int) -> ScheduleRecur up to times more times, without waiting.
E.Schedule.spaced
def spaced(delay: timedelta | None = None, **parts: Unpack[Parts]) -> ScheduleRecur 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],
) -> ScheduleRecur 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) -> ScheduleScale 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: timedeltaNot 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: floatNot 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() -> CurrentLoggersE.MinimumLogLevel
@final
@dataclass(frozen=True)
class MinimumLogLevel(ImplicitRequirement):
level: LogLevelE.MinimumLogLevel.default
def default() -> MinimumLogLevelE.CurrentLogLevel
@final
@dataclass(frozen=True)
class CurrentLogLevel(ImplicitRequirement):
level: SeverityThe level a bare log(...) call logs at.
E.CurrentLogLevel.default
def default() -> CurrentLogLevelE.CurrentLogAnnotations
@final
@dataclass(frozen=True)
class CurrentLogAnnotations(ImplicitRequirement):
annotations: Mapping[str, object]E.CurrentLogAnnotations.default
def default() -> CurrentLogAnnotationsE.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) -> strE.pretty_logger
pretty_logger: EffectonLoggerPath
E.Path
@final
@dataclass(frozen=True, init=False, repr=False)
class Path:
...E.Path.parent
parent: PathThe 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: strE.Path.suffix
suffix: strE.Path.stem
stem: strE.Path.parts
parts: tuple[str, ...]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() -> NeverE.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() -> ProtocolE.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 = 0A 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: PathE.FileSystem.PermissionDenied
@final
@dataclass(frozen=True)
class PermissionDenied(EffectonError):
path: PathE.FileSystem.PathIsADirectory
@final
@dataclass(frozen=True)
class PathIsADirectory(EffectonError):
path: PathE.FileSystem.PathIsNotADirectory
@final
@dataclass(frozen=True)
class PathIsNotADirectory(EffectonError):
path: PathE.FileSystem.PathAlreadyExists
@final
@dataclass(frozen=True)
class PathAlreadyExists(EffectonError):
path: PathE.FileSystem.DirectoryNotEmpty
@final
@dataclass(frozen=True)
class DirectoryNotEmpty(EffectonError):
path: PathE.FileSystem.FileSystemError
type FileSystemError = FileNotFound | PermissionDenied | PathIsADirectory | PathIsNotADirectory | PathAlreadyExists | DirectoryNotEmptyE.FileSystem.StatError
type StatError = FileNotFound | PermissionDeniedE.FileSystem.ReadError
type ReadError = FileNotFound | PermissionDenied | PathIsADirectoryE.FileSystem.WriteError
type WriteError = FileNotFound | PermissionDenied | PathIsADirectory | PathIsNotADirectoryE.FileSystem.MakeDirectoryError
type MakeDirectoryError = FileNotFound | PermissionDenied | PathAlreadyExists | PathIsNotADirectoryE.FileSystem.ReadDirectoryError
type ReadDirectoryError = FileNotFound | PermissionDenied | PathIsNotADirectoryE.FileSystem.RemoveError
type RemoveError = FileNotFound | PermissionDenied | DirectoryNotEmptyE.FileSystem.RenameError
type RenameError = FileNotFound | PermissionDenied | PathIsADirectory | PathIsNotADirectory | DirectoryNotEmptyE.FileSystem.CopyFileError
type CopyFileError = FileNotFound | PermissionDenied | PathIsADirectoryE.FileSystem.SymlinkError
type SymlinkError = FileNotFound | PermissionDenied | PathAlreadyExists | PathIsNotADirectoryE.FileSystem.FileType
type FileType = Literal['file', 'directory', 'symlink', 'other']E.FileSystem.FileInfo
@final
@dataclass(frozen=True)
class FileInfo:
type: FileType
size: int
modified_at: datetimeWhat 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: strThe 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: intE.HttpClient.InvalidJson
@final
@dataclass(frozen=True)
class InvalidJson(EffectonError):
method: Method
url: str
message: strE.HttpClient.HttpClientError
type HttpClientError = TransportError | StatusError | InvalidJsonE.HttpClient.Request
@final
@dataclass(frozen=True)
class Request:
method: Method
url: str
headers: Mapping[str, str] = field(default_factory=dict)
body: bytes | None = NoneWhat 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: boolE.HttpClient.Response.text
text: strThe 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 and the home directory; 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 both directories 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.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 = _HOMETracer
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: datetimeE.Tracer.Ended
@final
@dataclass(frozen=True)
class Ended:
start_time: datetime
end_time: datetime
exit: Exit[Any, Any]E.Tracer.SpanStatus
type SpanStatus = Started | EndedE.Tracer.Span
@runtime_checkable
class Span(typing.Protocol):
name: str
trace_id: str
span_id: str
parent: Span | None
kind: SpanKind
status: SpanStatus
attributes: Mapping[str, object]One named, timed unit of a program, as the Tracer represents it.
Ids are lowercase hex: 32 characters for the trace, 16 for the span. Spans in one tree share the trace id and chain through parent.
E.Tracer.Span.end
def end(end_time: datetime, exit: Exit[Any, Any]) -> Effect[None]E.Tracer.Span.attribute
def attribute(key: str, value: object) -> Effect[None]E.Tracer.NativeSpan
@final
@dataclass(eq=False)
class NativeSpan(Span):
name: str
trace_id: str
span_id: str
parent: Span | None
kind: SpanKind
status: SpanStatus
attributes: Mapping[str, object]The in-memory span both bundled tracers open; compares by identity.
E.Tracer.NativeSpan.end
def end(end_time: datetime, exit: Exit[Any, Any]) -> Effect[None]E.Tracer.NativeSpan.attribute
def attribute(key: str, value: object) -> Effect[None]E.Tracer.ParentSpan
@final
@dataclass(frozen=True)
class ParentSpan(ImplicitRequirement):
span: Span | NoneThe span new spans are opened under; None outside any span.
E.Tracer.ParentSpan.default
def default() -> ParentSpanE.Tracer.Protocol
@runtime_checkable
class Protocol(ImplicitRequirement, typing.Protocol):
...E.Tracer.Protocol.span
def span(
name: str,
*,
parent: Span | None,
start_time: datetime,
kind: SpanKind,
attributes: Mapping[str, object],
) -> Effect[Span]Open a span; with_span ends it once the wrapped effect settles.
E.Tracer.Protocol.default
def default() -> ProtocolE.Tracer.Live
@final
@dataclass(frozen=True)
class Live(Protocol):
...Opens in-memory spans that nothing collects.
E.Tracer.Test
@final
@dataclass
class Test(Protocol):
spans: list[Span] = field(default_factory=list, compare=False)Opens the same in-memory spans and records each one in spans.