fix(streaming): support bytes chunks + fix async DB access in generators (#1236, #1219)

[sansyrox]

Fixes two related StreamingResponse bugs that came out of user reports.

#1236 — yielding bytes crashed

Streaming binary data (yield b"...") raised 'bytes' object cannot be converted to 'PyString' because the Rust stream driver only extracted str. It now downcasts PyBytes first (used as-is) and falls back to UTF-8-encoding str, so file downloads / application/octet-stream work. A value that's neither str nor bytes is logged and ends the stream instead of being silently dropped.

@app.get("/download")
def download(request):
    def gen():
        with open(path, "rb") as f:
            while chunk := f.read(8192):
                yield chunk
    return StreamingResponse(gen(), media_type="application/octet-stream")

#1219 — awaiting async DB inside a generator crashed

await session.execute(...) (async SQLAlchemy) inside a streaming generator raised Task ... attached to a different loop. AsyncGeneratorWrapper drove the generator on a freshly-created event loop and swallowed every error (print + raise StopIteration), so the stream silently truncated.

It now drives the generator on the loop that was running when the StreamingResponse was constructed — i.e. the handler's loop, where the async resources are bound — via run_coroutine_threadsafe, falling back to a dedicated background loop only for sync handlers. Generator errors now propagate (and show up in the server logs) instead of disappearing.

Tests

• 4 unit tests (unit_tests/test_streaming_response.py): generator runs on the constructing loop, errors propagate, background-loop fallback for sync handlers, bytes pass-through.
• 2 integration tests (integration_tests/test_binary_streaming.py): sync and async bytes streaming.
• Existing SSE suite still green (the wrapper rewrite doesn't regress async SSE).

ruff / ruff format / isort / cargo fmt --check all clean.

Supersedes #1308 (which fixed only the bytes case) — happy to close that in favour of this.

Closes #1236
Closes #1219

🤖 Generated with Claude Code

Summary by CodeRabbit

• New Features
  • Streaming responses now support yielding raw bytes (including async streaming) for binary downloads.
  • Added binary streaming endpoints for both sync and async byte streams.
• Documentation
  • Updated Server-Sent Events docs with examples for streaming arbitrary raw bytes via StreamingResponse.
• Bug Fixes
  • Async streaming now propagates exceptions to the consumer instead of silently terminating.
• Tests
  • Added integration tests for sync/async byte streaming and unit tests covering async wrapper behavior and cleanup.

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
robyn	Ready	Preview, Comment	Jun 21, 2026 1:03am

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 5733b9ae-8e53-4753-a4f3-63c203f2413b

📥 Commits

Reviewing files that changed from the base of the PR and between 4006945 and bd0cf84.

📒 Files selected for processing (2)

• robyn/responses.py
• unit_tests/test_streaming_response.py

🚧 Files skipped from review as they are similar to previous changes (2)

• unit_tests/test_streaming_response.py
• robyn/responses.py

---

📝 Walkthrough

Walkthrough

This PR enables raw binary streaming in Robyn's response pipeline. The Rust stream handler gains a PyBytes-or-str dispatch path. The Python AsyncGeneratorWrapper is reworked with event-loop affinity and thread backing so async generators can be driven from sync iteration contexts. Public content types in StreamingResponse and SSEResponse are widened to accept str | bytes. New /stream/bytes and /stream/bytes_async routes, unit tests for the wrapper, integration tests for both endpoints, and documentation with examples are added.

Changes

Binary Streaming Support

Layer / File(s)	Summary
Rust stream: accept PyBytes or str from generators src/types/response.rs	create_python_stream now branches on PyBytes vs. String for each yielded value, encoding to bytes accordingly, logging an error and terminating the stream for any other type, then wraps into Bytes::from before yielding.
AsyncGeneratorWrapper: loop affinity, thread backing, bytes types robyn/responses.py	Replaces the prior wrapper with a loop-aware implementation that reuses the running loop from an async handler or creates a daemon background thread with a new event loop. Iteration uses asyncio.run_coroutine_threadsafe; StopAsyncIteration becomes StopIteration and BaseException propagates instead of being swallowed. Content types in StreamingResponse and SSEResponse are widened from str to str | bytes.
Binary streaming routes and comprehensive tests integration_tests/base_routes.py, unit_tests/test_streaming_response.py, integration_tests/test_binary_streaming.py	Adds /stream/bytes and /stream/bytes_async routes yielding three fixed byte chunks via sync and async generators. Unit tests cover AsyncGeneratorWrapper loop affinity, error propagation, background-loop ownership, and bytes pass-through. Integration tests assert status 200, Content-Type, and body equality.
Binary streaming documentation with examples docs_src/src/pages/documentation/en/api_reference/server_sent_events.mdx	Documents raw binary streaming patterns using StreamingResponse for file downloads and generated content. Includes a synchronous chunked download example with Headers and Content-Disposition, and an async generator example streaming CSV rows as encoded bytes.

Sequence Diagram(s)

sequenceDiagram
    rect rgba(70, 130, 180, 0.5)
        Note over Client,Actix: Binary Streaming Request Flow
    end
    participant Client
    participant Actix as Actix-Web Handler
    participant PyGen as Python Generator
    participant Wrapper as AsyncGeneratorWrapper
    participant RustStream as create_python_stream

    Client->>Actix: GET /stream/bytes or /stream/bytes_async
    Actix->>RustStream: start streaming
    RustStream->>PyGen: __next__() or via Wrapper
    alt async generator
        PyGen->>Wrapper: Wrapper.__next__
        Wrapper->>Wrapper: schedule __anext__ on loop
        Wrapper-->>PyGen: bytes or str chunk
    else sync generator
        PyGen-->>RustStream: bytes or str chunk
    end
    RustStream->>RustStream: PyBytes? → raw bytes / String? → UTF-8 bytes
    RustStream-->>Client: chunk in octet-stream response

Loading

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Poem

🐇 Hop hop, the bytes now flow,
No longer strings from head to toe,
The wrapper loops with threaded grace,
Async generators keep their place,
Big binary files stream strong and free —
What a lovely patch, says me! 🎉

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 41.38% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly identifies the two main bug fixes (bytes chunk support and async DB access) with specific issue numbers, accurately reflecting the changeset.
Description check	✅ Passed	The description comprehensively covers both issues, explains the root causes and fixes, includes test coverage details, and follows the template structure.
Linked Issues check	✅ Passed	All code changes directly address the linked issues: #1236 (bytes streaming) and #1219 (async DB access in generators) with proper fixes and test coverage.
Out of Scope Changes check	✅ Passed	All changes are directly related to fixing streaming response issues with bytes and async database operations; no unrelated modifications detected.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch fix/streaming-bytes

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (2)

unit_tests/test_streaming_response.py (1)

51-62: ⚡ Quick win

Add a cleanup assertion for the owned background loop

This test validates fallback behavior, but not loop/thread teardown. Please also assert the owned thread stops after exhaustion to guard against thread-leak regressions.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@unit_tests/test_streaming_response.py` around lines 51 - 62, After the
assertion that validates the wrapper yields the expected values in the
test_wrapper_without_running_loop_uses_background_loop function, add an
additional assertion to verify that the background loop/thread owned by
AsyncGeneratorWrapper (indicated by _owns_loop being True) is properly cleaned
up and stopped after the generator is exhausted. This ensures the wrapper does
not leak the background thread resource after the generator completes iteration.

integration_tests/test_binary_streaming.py (1)

17-21: ⚡ Quick win

Assert Content-Type for the async bytes endpoint too

The sync case checks header contract; the async case should mirror it to catch regression in response metadata.

Test delta

 def test_stream_bytes_async(session):
@@
     r = requests.get(f"{BASE_URL}/stream/bytes_async", timeout=TIMEOUT)
     assert r.status_code == 200
+    assert r.headers.get("Content-Type") == "application/octet-stream"
     assert r.content == EXPECTED

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@integration_tests/test_binary_streaming.py` around lines 17 - 21, The
test_stream_bytes_async function is missing a Content-Type header assertion to
match the sync case (test_stream_bytes) and ensure the response metadata
contract is maintained. Add an assert statement to verify the Content-Type
header value in the response object r.headers after the existing status_code
assertion, using the same expected content type that the sync test validates.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@robyn/responses.py`:
- Around line 95-96: The daemon thread created by self._thread at lines 95-96 is
only stopped via the _finish() method, but if streaming terminates early before
the wrapper is fully exhausted, _finish() may never be called, leaving the
thread alive. Ensure the thread is properly cleaned up by implementing resource
management that guarantees _finish() is called regardless of how streaming ends,
such as using a context manager pattern with __enter__ and __exit__ methods, or
implementing __del__ to clean up the _thread, or wrapping the streaming logic
with try/finally to ensure cleanup happens on all code paths.

---

Nitpick comments:
In `@integration_tests/test_binary_streaming.py`:
- Around line 17-21: The test_stream_bytes_async function is missing a
Content-Type header assertion to match the sync case (test_stream_bytes) and
ensure the response metadata contract is maintained. Add an assert statement to
verify the Content-Type header value in the response object r.headers after the
existing status_code assertion, using the same expected content type that the
sync test validates.

In `@unit_tests/test_streaming_response.py`:
- Around line 51-62: After the assertion that validates the wrapper yields the
expected values in the test_wrapper_without_running_loop_uses_background_loop
function, add an additional assertion to verify that the background loop/thread
owned by AsyncGeneratorWrapper (indicated by _owns_loop being True) is properly
cleaned up and stopped after the generator is exhausted. This ensures the
wrapper does not leak the background thread resource after the generator
completes iteration.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: a6c75185-22ea-4dae-9fc1-e33ad77ba86d

📥 Commits

Reviewing files that changed from the base of the PR and between 86f7695 and 7114d70.

📒 Files selected for processing (5)

• integration_tests/base_routes.py
• integration_tests/test_binary_streaming.py
• robyn/responses.py
• src/types/response.rs
• unit_tests/test_streaming_response.py

[codspeed-hq]

Merging this PR will not alter performance

✅ 209 untouched benchmarks

---

_{Comparing fix/streaming-bytes (bd0cf84) with main (86f7695)}

[sansyrox]

Good catch on the background-loop thread leak — fixed in bd0cf84.

• The thread target and cleanup are now staticmethods that take the loop as an argument rather than bound methods, so they don't keep the wrapper alive (a bound-method target=self._run_loop would have created a self-reference that prevents GC, which would have defeated any GC-based cleanup).
• A weakref.finalize stops the loop when the wrapper is dropped, so the loop closes and the daemon thread exits even if the stream ends early (client disconnect, or an unsupported chunk type). _finish() still calls it for the normal path.

Added a unit test that abandons the stream after one chunk and asserts the background thread is joined.

Also verified the fix end-to-end against a live server: streaming a generator that awaits a future bound to the handler's loop returns correctly on this branch, and returns an empty body on main (the original #1219 failure).