without¶
Core interfaces and executor for without: the sans-IO stream-processor substrate.
Bases: Protocol
A stream viewed as its latest value: the "behavior" half of the model.
Where consuming a stream sees every event, current samples the latest
and never blocks. This is how long-lived state (config, a connection pool) is
read: a context is just another processor's output that a reader samples
rather than consumes. current MUST return a value; a context is never
"not ready". The reader only ever gets a value, never a writable place.
Bases: Protocol
A transformation from a stream of inputs to a stream of outputs.
This is the only thing a user writes, and the only node type: a processor's output stream becomes another processor's input stream, all the way down.
I/O is decoupled, not forbidden. A processor MAY await I/O while
handling an event (a database query, a closed-lifespan sub-request), reading
its dependencies from injected Context values; this is why a scan's
step is async. The point is not to ban I/O but to separate it into the
right abstractions so the parts stay reusable: sources at the edge, behaviors
via sample, effects contained in the step. The one rule: an effect MUST
NOT escape the entrypoint. A processor awaits its I/O to completion and MUST
NOT hand a half-open resource (an open socket, an unfinished task it does not
own) back to the runtime. Testing injects fake Context dependencies.
Bases: Protocol
An asynchronous sequence of values.
A stream is the single shape every connection has. Sources that touch the outside world (a socket, a file watcher, a clock) are streams too: a stream is just the one shape every connection takes, whoever does the I/O.
dataclass
¶
The result of folding one event into a scan's state.
A value, never a place: a step returns the next state and the single
output it emits, and mutates nothing the caller can observe. Splitting
one event into several outputs is a wiring-style concern, not a per-step
one, so a transition carries one output rather than a collection.
dataclass
¶
async
¶
Wait for the drain to publish the next value, then return it.
The deterministic counterpart to current on the behavior edge: where
current reads the latest value and never blocks, updated blocks until
the background drain consumes and publishes the next value from the
source, then returns it. It is the "await next update" signal a reader
waits on (a test asserting on post-reload state, a control loop reacting
to a config change) instead of guessing how long the background task
needs. If the source raises instead of yielding, the wait raises that
error rather than hanging, and the failure is terminal: once the source
has failed, every later call re-raises it rather than registering a
waiter that can never resolve. If the context closes first, the wait is
cancelled.
Each call registers its own one-shot future resolved by the next publish,
so concurrent waiters are independent: cancelling one deregisters it at
once and never disturbs another. Like current, it inherits latest-wins: a waiter sees only
publishes after it starts waiting, and a source that publishes faster
than the reader re-arms collapses the values it missed. So updated is a
"the state has moved on" signal, not a way to observe every value;
consume the stream for that.
Build a processor that drops events for which reject is true and keeps the rest.
from_selector with the opposite polarity: where a selector keeps the
matching subset, a filter removes it (filtering those events out). The two
are duals, from_filter(reject) being from_selector of the negated
predicate, and both exist because naming the intent positively at the call
site (from_filter(is_health_check), from_selector(is_error)) reads better
than threading a negation through a predicate. Note this is the opposite
polarity to Python's built-in filter, which keeps matches, as
from_selector does; from_filter is itertools.filterfalse. Like
from_selector, the predicate is async.
Build a leaf that folds a stream of events into a single final state.
The stateful terminus, dual to from_scan: where from_scan threads
state and emits an output every step (a scan), from_fold threads state
and yields only the final accumulated value when the stream ends (a true
reduce). The step MAY await contained I/O, so a fold whose result you
ignore is also how you run a stateful consumer for its effects.
Build a processor from a stateless step: each event maps to one output.
The counterpart to from_scan for a processor that holds no state.
Each event is handled independently of every other, so there is no
initial to seed and no Transition to thread: step maps an event
straight to its single output. Like from_scan the step is async
so it MAY await contained I/O, and the effect MUST complete within each
call (see Processor). Splitting one event into several outputs is a
separate, wiring-style concern, not a per-step one, so the step returns a
single value rather than a collection.
from_scan(
initial: S,
step: Callable[
[In, S], Awaitable[Transition[S, Out]]
],
) -> Processor[In, Out]
Build a processor from a stateful step that emits an output every event.
step is the kernel: given an event and the current state it returns the
next state and the output it emits. It is async so it MAY await contained I/O
(reading dependencies from Context values captured by closure), but a
step that does no I/O is just an async def that never awaits.
from_scan supplies the loop that threads state across the input stream,
emitting one output per event: a scan, not a reduce (the collapse-to-one-
value form is from_fold). The effect MUST complete within each call (see
Processor).
Build a processor that keeps events for which keep is true and drops the rest.
A Processor is any Stream -> Stream function, so a step is under no
obligation to emit exactly one output per event the way from_map does: it
MAY yield zero. A selector is that zero-or-one case, an async generator that
re-emits an event on a match and skips to the next input otherwise, keeping
the matching subset through unchanged. This is the same sense as Python's
built-in filter (keep the matches).
Like every other builder step, keep is async: a predicate is one color
of function throughout without, so a decision that needs to await I/O (an
async permission check, a lookup) composes without ceremony, and a pure
decision simply never awaits (it MAY read injected Context values, whose
current never blocks). The polarity-opposite builder is from_filter,
which drops the matches. Emitting several outputs per event, by contrast, is
a wiring-style concern rather than a builder one (see without.wiring).
Build a leaf that consumes a stream for its effects and emits nothing.
The stateless terminus, dual to from_map: where a map turns each event
into an output, a sink turns each event into an effect and yields no output
stream at all. Awaiting it drains the stream to completion (or runs forever,
for an unbounded source driven inside a background_task). The step
MAY await contained I/O.
async
¶
as_async_iterator(
items: AsyncIterable[T] | Iterable[T],
) -> AsyncIterator[T]
Normalize a sync or async iterable into a single async iterator.
Lets code that consumes via async for/anext accept either kind without
branching on the iteration protocol at every use.
async
¶
background_task(
coro: Coroutine[object, object, T],
) -> AsyncIterator[Task[T]]
Run coro as a task for the duration of the with block.
The task is started on entry and cancelled (then awaited) on exit, so it is bounded by the block and never leaks. If it finishes on its own with an exception, that surfaces when the block exits.
async
¶
Cancel every future, then await them all so their teardown completes.
Two phases on purpose: cancelling the whole set before awaiting any of them
lets them tear down concurrently, instead of serially cancelling and waiting
for one at a time. None entries are skipped, so a caller holding an optional
task (task: asyncio.Task | None) can pass it without a guard. The futures are
materialized first, so a caller may pass a live set the awaits will mutate. Each
future's own CancelledError is suppressed; any other exception it raises during
teardown propagates, which also ends the loop, so the futures behind it in the set
are cancelled but never awaited. That is the right order for a set of pending
futures, where nothing has an exception to raise; pass one that may already hold a
failure and the ones behind it lose their teardown to it. Filter to what is still
running when the set can contain both.
async
¶
limit_concurrency(
aws: AsyncIterable[Awaitable[T]]
| Iterable[Awaitable[T]],
limit: int,
) -> AsyncIterator[Future[T]]
Run awaitables from aws with at most limit in flight, yielding each as it finishes.
A bounded-concurrency driver: it pulls the next awaitable from aws only
while fewer than limit are already running. So a lazy source (an async
generator that produces each unit of work on demand) is never advanced past
the limit. That is what lets it gate a side-effecting source: an accept loop
whose generator awaits socket.accept() only when pulled will never accept
more connections than it can serve.
Each completed awaitable is yielded as a Future; call .result() on it to
read the value or re-raise its exception. On early exit or cancellation, any
still-running awaitables are cancelled and awaited, so none outlive the
iteration.
limit must be at least 1; a non-positive limit is a ValueError, since it
could only ever stall the source rather than run it.
Adapted from Limiting concurrency in asyncio.
async
¶
Suspend the current task until it is cancelled.
The idiom for a coroutine whose job is to stay alive until its surrounding scope tears it down: a server's run loop holding a bound socket open, a process that should idle until signalled. It awaits a future that never resolves, so it consumes nothing and ends only on cancellation.
async
¶
timeout(duration: timedelta | None) -> AsyncIterator[None]
Bound the with block by duration, or leave it unbounded when None.
A timedelta-typed, nullable wrapper over asyncio.timeout: None disables the
bound (an always-open context), and a duration raises TimeoutError if the block
outlives it. Modelling "no limit" as None keeps that choice a first-class value at
the call site, rather than a sentinel float threaded through the same parameter.
async
¶
Drain a Stream into a list: the terminal that materializes every value.
The dual of stream_from_iterable. It runs until the source ends, so it suits bounded
streams (a finished request, a shut-down queue); an endless source never
returns.
Compose two processors on the event edge: first then second.
The join type B may differ from A and C, so this adapts as well as
chains. Pure composition (the only event-edge connector that needs nothing
running); nest for three or more stages. When second is a Sink rather than
a Processor the result is a Sink too: the same wiring, terminated, which is
how a middleware chain (a filter, an enrichment) is prefixed onto a terminal
consumer such as a writer.
async
¶
sample(source: Stream[T]) -> AsyncIterator[Sample[T]]
Connect to a stream on the behavior edge: read its latest value, not each.
The first value is sampled eagerly, so the context is never "not ready". A
background task keeps the held value current while the with block is open,
dropping intermediate values (latest-wins, no backpressure). A reader reads
the held value through current (latest, non-blocking) or waits for the next
one through updated (the deterministic "await next update" signal); the held
value is mutated only by the drain. The yielded Sample is a Context, so a
caller that only reads current can treat it as one. When the block exits,
any still-pending updated waits are cancelled, so a task awaiting one is not
left hanging on a context that has closed.
async
¶
spool(source: Stream[T], ahead: int) -> AsyncIterator[T]
Drive a source ahead of its consumer through a bounded queue: read-ahead.
A background task pulls from source as fast as backpressure allows and
drops each value into a queue of at most ahead items; the returned stream
yields from that queue. So the source is driven independently of how fast
the consumer pulls: a pull-based producer (an accept loop, a file's chunks, a
DAG's executed iterator) keeps making progress while a slower consumer
catches up, up to ahead items of slack before put blocks and backpressure
reaches the producer. That overlaps the producer's work with the consumer's,
e.g. reading the next file chunk while the current one is still being written
to a socket.
ahead must be at least 1: the bound is the backpressure, so an unbounded
spool (which could let a fast producer grow memory without limit) is a
ValueError rather than a silent default. When source ends the queue is
shut down and the stream ends once drained; if source raises, the spooled
items still drain and then the error surfaces. Closing the stream early
cancels the background task, so the producer never outlives its consumer.
Compose middleware into one, first argument outermost; stack() is identity.
A middleware is (handler, *context) -> handler: it wraps a handler, given some
fixed context, into a new handler of the same type. The context is whatever the
setting threads through unchanged: nothing for a client exchange (Endo[H]), the
connection state and scope for a server handler. stack threads the same context
into every middleware and chains the handler through them, first outermost, so
stack(f, g)(handler, *context) is f(g(handler, *context), *context).
Generic over the handler H (the value each middleware transforms) and the context
pack *Ctx, which is bound once per call: every middleware in one stack(...) must
therefore share a shape, and mixing shapes is a type error. The pack passes through
untouched (never wrapped element-wise), which is exactly why one variadic generic
covers every arity here where a heterogeneous ladder would be needed.
async
¶
stream_from_iterable(
values: Iterable[T],
) -> AsyncIterator[T]
Expose a fixed iterable as a Stream: the simplest source.
Turns already-in-hand values into the pull-based Stream the rest of
without consumes, e.g. to emit a fixed reply or to feed a processor under
test. stream_from_queue is the push-source counterpart.
async
¶
stream_from_queue(queue: Queue[T]) -> AsyncIterator[T]
Expose a queue as a Stream: the bridge from a push source to a pull stream.
A source that pushes (a server's accept loop, a callback-based client, a
pub/sub subscriber) drops values into a queue; this turns that queue into the
pull-based Stream the rest of without consumes. It ends gracefully when
the queue is shut down (queue.shutdown()): remaining items still drain, then
get raises QueueShutDown and the stream ends, letting a downstream fold
return its final value. Shutting the queue down is thus the closable-stream
signal; without it the stream never ends on its own and must be driven inside
a background_task or otherwise cancelled by its consumer.
Fan one stream out to every sink: the terminal counterpart to compose.
Where compose chains a processor onto a single sink, tee splits the stream
across several, after the Unix tool that writes one input to many destinations.
Every sink sees every event, in order, and the input is consumed exactly once.
The caller controls the split point purely by placement, since each argument is
itself a Sink: whatever is composed before the tee is the shared prefix
(parsed and enriched once), and each branch is its own Sink, carrying its own
filtering, rendering, and terminal. A branch MAY itself be another tee, so
"one input, several sink groups" nests without new machinery.
One pump reads the source once and pushes each value onto every branch's bounded
queue; each sink drains its own queue concurrently, and when the source ends the
queues are shut so each branch's stream ends and its sink returns. Every branch
MUST be consumed to completion and concurrently: a sink that stops early leaves
its queue to fill and stalls the pump (real sinks, a filter that drops events
included, drain every input). A sink failure tears the whole tee down and
surfaces as an ExceptionGroup, so a broken branch fails loud rather than
silently starving the rest.
buffer is how far a branch may run ahead: the queues are bounded to it, so the
slowest branch gates the pump (and thus backpressure onto the source), while a
larger value trades memory for slack so a fast branch need not wait on a slow
one. The default 1 keeps memory O(sinks). At least one sink is REQUIRED, and
buffer MUST be at least 1 (an unbounded branch could grow memory without limit).
async
¶
ticks(
every: timedelta,
*,
now: Callable[[], datetime] = utc_now,
) -> AsyncGenerator[datetime]
A Stream of moments, one now and one every every after: the clock as a source.
The source periodic work runs off, so that when something happens is a stream a
caller supplies rather than a loop inside the thing being done. A cache sweep, a
config refresh, a queue's housekeeping: each becomes a Sink that says only what
happens per event, and composing it with this says how often. A while True with a
sleep in it can only ever be a timer, and it buries the schedule inside the work;
a sink over a stream runs off this, off stream_from_queue when an operator pokes
it, or off stream_from_iterable in a test that chooses the instants.
Each tick carries its moment, so a consumer needs no clock of its own and a test controls time by choosing values rather than by patching one.
It yields before it sleeps, so the first event lands at once rather than one interval
later, and it never ends on its own: drive it inside a background_task, a task
group, or anything else that will cancel it.
The sleep goes after the yield rather than being measured from it, so the period is
every plus however long the consumer took, and the moments drift later by that much
each time. That is the right trade for the work this drives: a sweep that runs on a
fixed period instead of a fixed cadence can never overlap itself, where a scheduler
that chased a wall-clock grid would fire back-to-back to catch up after one slow pass,
which is exactly the wrong response to a dependency that has gone slow. What it means
for a caller is that every is a floor on the gap between events rather than a
promise about when each one lands, so a consumer that needs the true elapsed time
reads the moment it is handed rather than counting ticks.
An interval that is not positive is refused rather than run, for the same reason
drive refuses a limit below one: it has no sensible reading, and taken literally
it is a loop that yields as fast as the sink can consume, which pins a core to do
housekeeping. A zero arrives from a configured duration whose setting was never set,
so it is worth one comparison here.