feat(streaming): add EventStreamer for iterating stream events - #1567
feat(streaming): add EventStreamer for iterating stream events#1567ajbozarth wants to merge 1 commit into
Conversation
ajbozarth
added
area/stdlib
|
@jakelorocco here's my initial draft POC for a events iterator for streaming. This is the better of the two POCs I wrote and I'll leave this here for your review while I'm out. Once I get back next week we can sync and I'll continue work on this |
jakelorocco
left a comment
jakelorocco
left a comment
There was a problem hiding this comment.
This pretty much looks like exactly what I was looking for; thank you!
7d03c9a to
52987f8
Compare
stream(as_events=True) returns an EventStreamer that yields a stream's typed StreamEvent objects via async-for, instead of the Streamer's str chunks. It runs the stream to completion on a background task and delivers events through a queue, so the terminal CompletedEvent and ErrorEvent reach the consumer through iteration, staying queued behind an early break until it resumes — which iterating the driver's generator cannot guarantee. Additive and opt-in: the default stream() chunk contract is unchanged. Events are threaded to an optional per-stream queue in _emit_event; the STREAMING_EVENT hook still fires on all paths, so telemetry is unaffected. stream() is split into _stream() plus a dispatching wrapper with typed overloads on as_events. Assisted-by: Claude Code Signed-off-by: Alex Bozarth <ajbozart@us.ibm.com>
52987f8 to
81f392a
Compare
|
PR is updated and moved out of draft POC, ready for review |
planetf1
left a comment
planetf1
left a comment
There was a problem hiding this comment.
I found three correctness and documentation issues inline.
| if isinstance(exc, Exception): | ||
| self._ready.set_exception(exc) | ||
| else: | ||
| self._ready.cancel() |
There was a problem hiding this comment.
Could this preserve non-cancellation BaseExceptions rather than cancelling _ready? If _stream() raises a custom BaseException, KeyboardInterrupt, or SystemExit during setup, awaiting stream(..., as_events=True) receives CancelledError instead. The caller cannot see or handle the real failure, even though the default stream() path preserves it; stream() then retrieves the original pump exception only to discard it. I suggest considering reserving _ready.cancel() for asyncio.CancelledError, using _ready.set_exception(exc) for other exceptions, and updating the regression test to expect the original exception.
|
|
||
| def __init__(self) -> None: | ||
| """Create an idle handle; `stream()` spawns the pump task.""" | ||
| self._queue: asyncio.Queue[StreamEvent | None] = asyncio.Queue() |
There was a problem hiding this comment.
This queue is unbounded while the pump drains the backend independently of the consumer. A slow or stalled consumer retains every event in memory; with chunking=None, that includes one ChunkEvent per raw delta. A sufficiently long response can therefore grow the process's memory until it is killed, rather than applying backpressure to generation. Could we define a bounded, cancellation-safe buffering policy here? A simple maxsize needs a close-path test too, so terminal events and teardown cannot block when the consumer has stopped.
| ) as streamer: | ||
| async for event in streamer: | ||
| match event: | ||
| case ChunkEvent(): |
There was a problem hiding this comment.
The preceding example imports only stream, so a reader who follows it gets NameError on the first ChunkEvent (and likewise for the other event classes). Could you consider importing the event types in the preceding import block or directly before this example?
3 participants
Issue
Follow-up to the design discussion on #1543 (now merged) — streaming event consumption ergonomics.
Description
Adds an opt-in
EventStreamer, returned bystream(..., as_events=True), for iterating a single stream's typedStreamEventobjects directly withasync for. The only other route to those events is the process-globalSTREAMING_EVENThook, which is awkward for a single stream — you register a plugin and demultiplex onstreaming_id.Under the hood it wraps a
Streamer, drives it to completion on a background task, and delivers the events through a queue the consumer drains withasync for.It is additive and opt-in:
stream()'s default chunk-iteratingStreameris unchanged, and theSTREAMING_EVENThook still fires on every stream, so telemetry and multi-stream observers are unaffected.EventStreameris just the ergonomic way to consume one stream's events inline; the hook remains the right tool for observing many at once.The streaming examples and docs go back to iterating events, reverting the
streaming_event-hook workaround #1543 adopted when the iterator was removed; the v0.7→v0.8 migration guide now mapsresult.events()tostream(as_events=True)to match.Testing
Unit tests cover
EventStreamer's event-mode paths (natural completion, requirement failure, error re-raise, early break), resume-and-drain after a break, teardown edge cases (idempotentaclose, a faulted/abandoned pump, aBaseExceptionduring setup that must not hang), and setup-error surfacing;test/typing/check_stream_as_events.pycovers theas_eventsoverloads including a dynamicbool.Attribution
Adding a new component, requirement, sampling strategy, or tool?
If your PR adds or modifies one of the types below, check the matching box. A checklist of type-specific review items will be posted as a comment.
NOTE: Please ensure you have an issue that has been acknowledged by a core contributor and routed you to open a pull request against this repository. Otherwise, please open an issue before continuing with this pull request.