without_streams

The sans-IO stream-processor substrate for without: streams, processors, contexts, and the wiring between them.

without_streams

Fold

Fold = Callable[[Stream[In]], Awaitable[S]]

Sink

Sink = Callable[[Stream[In]], Awaitable[None]]

Endo

Endo = Callable[[T], T]

Context

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.

current

current() -> T

Processor

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.

Stream

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.

Transition dataclass

Transition(state: S, output: Out)

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.

state instance-attribute

state: S

output instance-attribute

output: Out

Sample dataclass

Sample(_value: T)

current

current() -> T

updated async

updated() -> T

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.

from_filter

from_filter(
    reject: Callable[[T], Awaitable[bool]],
) -> Processor[T, T]

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.

from_fold

from_fold(
    initial: S, step: Callable[[In, S], Awaitable[S]]
) -> Fold[In, S]

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.

from_map

from_map(
    step: Callable[[In], Awaitable[Out]],
) -> Processor[In, Out]

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

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).

from_selector

from_selector(
    keep: Callable[[T], Awaitable[bool]],
) -> Processor[T, T]

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).

from_sink

from_sink(
    step: Callable[[In], Awaitable[None]],
) -> Sink[In]

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.

close_stream async

close_stream(source: Stream[object]) -> None

Close source if it is a generator: how a consumer releases a stream it abandons.

Stream is __aiter__-only, so a source may be a generator holding a finally (a file, a task, a connection) or an object with nothing to release. This is the difference between the two, in one place, because the alternative is every consumer that can stop early carrying the same check.

Reach for it wherever a consumer may not drain what it was given, which is the case whenever it can raise or be closed part-way through: without it the generator's cleanup waits on garbage collection, so a long-lived source outlives the consumer by an indeterminate amount. contextlib.aclosing is the same guarantee for a source already known to be a generator.

collect async

collect(source: Stream[T]) -> list[T]

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

compose(
    first: Processor[A, B], second: Processor[B, C]
) -> Processor[A, C]

compose(
    first: Processor[A, B], second: Sink[B]
) -> Sink[A]

compose(
    first: Processor[A, B],
    second: Processor[B, C] | Sink[B],
) -> Processor[A, C] | Sink[A]

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.

offload async

offload(
    work: Callable[[Iterator[list[T]]], None],
) -> AsyncIterator[Sink[T]]

Run a blocking work on a dedicated thread, fed by the yielded async Sink.

The async-to-sync direction, and the exact counterpart of stream_from_blocking: there a blocking source feeds an async consumer, here an async producer feeds a blocking consumer. Both put one long-lived thread on the blocking side and bridge with a queue, because the alternative (an async wrapper library hopping to a worker thread per operation) pays that round trip on every item. Here a single thread owns the resource and does all the I/O: work is plain blocking Python, and the yielded Sink drops each item onto a thread-safe queue the worker drains. No per-item thread hop, and the consumer body stays ordinary synchronous code.

The two are not mirror images, and the asymmetry is real rather than an oversight. A source has to be pulled, so stream_from_blocking bounds its queue and pushes backpressure into the producer; a sink is pushed, so the bound here would have to become a decision about what to do when the worker falls behind (block the async side, or drop). See the note on the queue below: this first cut takes neither, deliberately.

Items arrive in bursts: each element of the iterator is everything available on the queue at that instant (at least one item, blocking for the first). A burst boundary is therefore exactly the moment the worker has caught up, which is where a writer flushes: under load bursts are large and flushes are few, and when idle each burst is a single item flushed at once. So durability needs no flush-frequency knob; it falls out of the queue's own backlog.

Lifecycle is bounded by the with block: the thread starts on entry, and on exit the queue is shut down so the worker drains what remains and then ends, and the thread is joined (so a file is closed before the block returns). If work raises, that surfaces when the block exits.

Nest this outside the consumer that drives the sink, so the worker outlives the draining: async with offload(...) as writer: around whatever drives writer, not the other way round.

The queue is unbounded in this first cut: the async side never blocks or drops, at the cost of growing memory if the worker cannot keep up with a sustained burst (a stalled disk). A bounded, drop-counting variant is a deliberate follow-up. The bidirectional case (a full Processor bridged onto a thread, with an output queue as well) is intentionally out of scope; this covers the terminal-sink need (writing) without that extra state.

sample 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.

spool 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.

stack

stack(
    *middleware: Callable[[H, *Ctx], H],
) -> Callable[[H, *Ctx], H]

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.

stream_from_blocking async

stream_from_blocking(
    values: Iterable[T], *, ahead: int = 1
) -> AsyncIterator[T]

Expose a blocking iterable as a Stream, without stalling the event loop.

stream_from_iterable is the right adapter for values already in hand, but it pulls each one on the loop's own thread, so a source that blocks between items (a pipe, sys.stdin, a driver with no async client, a Queue.get) parks every other task while it waits. This runs the whole iteration on one worker thread instead and hands each value across a bounded queue, so the loop stays free and the producer runs ahead by at most ahead items before backpressure reaches it. It is spool's counterpart for a source that is not async in the first place.

Handing the whole loop to the thread, rather than awaiting one next at a time, is what makes it pipeline: the producer fetches the next value while the consumer is still working on the last, which per-item offloading cannot do. The cost is the thread being held for the source's whole lifetime rather than one read, which is the right trade for a long-lived source and the wrong one for many short ones.

Two consequences worth knowing, both from the fact that a blocked thread cannot be cancelled:

• Abandoning the stream stops the handover immediately, but the thread itself lives until its source produces one more item or ends. For sys.stdin that can be forever, so the worker is a daemon thread: the process exits without waiting for it, where a pooled thread would hang asyncio.run's shutdown.
• The producer's own cleanup runs when that thread unwinds, which may be after the consumer has moved on. A generator source is closed explicitly there rather than left to garbage collection, so its finally is as prompt as it can be, but a source that must release a resource on a deadline wants an async adapter rather than this one. Anything else is left open, because it belongs to whoever passed it in and may well be read again.

stream_from_iterable 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.

stream_from_queue 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.

tee

tee(*sinks: Sink[T], buffer: int = 1) -> Sink[T]

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).

ticks 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.