StreamTransformer

Return the projection dict.

Keys become entries in run.extensions. If the transformer has _native = True, keys are also set as direct attributes on the run stream.

StreamChannel instances in the return value are automatically wired by the StreamMux for protocol event auto-forwarding.

Handle an event on the sync lane.

Called for every event before it is appended to the main event log. Subclasses must override either process or aprocess. The default raises so a missing override fails loudly rather than silently passing every event through.

Handle an event on the async lane.

The mux awaits this before dispatching to the next transformer, so a slow aprocess serializes the pipeline. Use it only when a later transformer — or a consumer reading the event synchronously — must see the result of the async work (e.g. PII redaction that mutates event in place).

The default delegates to process, so purely-sync transformers run unchanged under astream().

Called when the run ends normally (sync lane).

Override to close StreamChannels, resolve promises, or perform other teardown. StreamChannel instances in the projection dict are auto-closed by the mux.

Called when the run ends normally (async lane).

By the time this runs, the mux has already awaited every task started via schedule(), so StreamChannels can be closed here without a last-task-wins race.

The default delegates to finalize.

Called when the run ends with an error (sync lane).

Override to fail StreamChannels, reject promises, or perform other teardown. StreamChannel instances in the projection dict are auto-failed by the mux.

Called when the run ends with an error (async lane).

The mux cancels and awaits every task started via schedule() before calling this, so cleanup doesn't race with in-flight work.

The default delegates to fail.

Schedule a coroutine tied to this transformer's lifecycle.

The mux holds the task reference, awaits all scheduled tasks during aclose() before calling afinalize(), and cancels them on afail(). Authors don't need to track tasks or implement the last-task-closes-the-log dance.

Requires a running event loop — call only under astream(). Set requires_async = True on the class so registration under sync stream() fails fast with a clear message.