Overview

Like Batman's gadgets streaming from the Batcave to his utility belt, Robyn provides built-in support for streaming responses. This allows you to send data in chunks, perfect for large files, real-time updates, and server-sent events.

Creating Streaming Responses

There are two ways to create streaming responses in Robyn:

1. Using the streaming parameter

Server

@app.get("/bat-signal", streaming=True)
async def stream_signal():
    async def signal_generator():
        while True:
            yield b"Bat-Signal Active\n"
            await asyncio.sleep(1)
    
    return Response(
        status_code=200,
        headers={"Content-Type": "text/plain"},
        description=signal_generator()
    )

Client

curl http://localhost:8000/bat-signal

2. Returning an Iterator/Generator

Robyn automatically detects iterators and generators and treats them as streaming responses:

Server

@app.get("/bat-signal")
async def stream_signal():
    async def signal_generator():
        while True:
            yield b"Bat-Signal Active\n"
            await asyncio.sleep(1)
    
    return signal_generator()  # Automatically detected as streaming

Response Object

The Response class supports streaming through its constructor parameters:

Response(
    status_code=200,
    headers={"Content-Type": "text/plain"},
    description=generator(),  # Can be str, bytes, or iterator/generator
    streaming=True  # Optional, automatically set for iterators/generators
)

Parameters

NameTypeDescriptionDefault
status_codeintResponse status code200
headersDict[str, str]Response headersNone
descriptionUnion[str, bytes, Iterator, AsyncIterator]Content to streamNone
streamingboolWhether to treat as streaming responseFalse

Supported Types

Like Batman's versatile arsenal, the streaming response system supports multiple data types:

Binary

# Raw binary data (like Batcomputer logs)
async def generator():
    yield b"Batcomputer Log Entry\n"

Text

# Text messages (like Alfred's updates)
async def generator():
    yield "Master Wayne, your tea is ready\n".encode()

Numbers

# Numbers (like Batmobile telemetry)
async def generator():
    yield str(speed).encode()

JSON

# JSON data (like Gotham City surveillance)
async def generator():
    yield json.dumps({"location": "Crime Alley"}).encode()

Server-Sent Events

For real-time updates from the Batcomputer:

Server

@app.get("/batcomputer/events", streaming=True)
async def batcomputer_feed():
    async def event_generator():
        while True:
            data = {
                "time": time.time(),
                "alerts": get_gotham_alerts()
            }
            yield f"data: {json.dumps(data)}\n\n".encode()
            await asyncio.sleep(1)
    
    return Response(
        status_code=200,
        headers={
            "Content-Type": "text/event-stream",
            "Cache-Control": "no-cache",
            "Connection": "keep-alive"
        },
        description=event_generator()
    )

Client

const evtSource = new EventSource("/batcomputer/events");
evtSource.onmessage = (event) => {
    console.log(JSON.parse(event.data));
};

File Downloads

For streaming large files from the Batcomputer archives:

Server

@app.get("/batcomputer/files", streaming=True)
async def download_files():
    async def file_generator():
        chunk_size = 8192  # Size of a Batarang
        with open("case_files.dat", "rb") as f:
            while chunk := f.read(chunk_size):
                yield chunk
    
    return Response(
        status_code=200,
        headers={
            "Content-Type": "application/octet-stream",
            "Content-Disposition": "attachment; filename=evidence.dat"
        },
        description=file_generator()
    )

Client

curl -O http://localhost:8000/batcomputer/files

Helper Functions

Robyn provides helper functions for common streaming scenarios:

from robyn import html

@app.get("/bat-report", streaming=True)
async def stream_html():
    async def generator():
        yield "<html><body>"
        for i in range(5):
            yield f"<p>Bat-Signal sighting {i}</p>"
            await asyncio.sleep(0.1)
        yield "</body></html>"
    
    return html(generator(), streaming=True)

Common Headers

Plain Text

headers = {"Content-Type": "text/plain"}

Server-Sent Events

headers = {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    "Connection": "keep-alive"
}

File Downloads

headers = {
    "Content-Type": "application/octet-stream",
    "Content-Disposition": "attachment; filename=file.dat"
}

Error Handling

Even Batman needs contingency plans:

async def generator():
    try:
        for item in evidence_items:
            yield process(item)
    except Exception as e:
        yield f"Alert: Batcomputer Error - {str(e)}".encode()
        return

Testing

Test your streaming responses like Batman testing his equipment:

@pytest.mark.asyncio
async def test_bat_signal():
    async with aiohttp.ClientSession() as client:
        async with client.get("http://localhost:8080/bat-signal") as response:
            chunks = []
            async for chunk in response.content:
                chunks.append(chunk.decode())
            assert len(chunks) > 0

Best Practices

Encode Data

Always encode strings to bytes (like encrypting Bat-communications)

Chunk Size

Use appropriate chunk sizes (8KB-64KB for efficient data transfer)

Resource Management

Clean up resources (like Batman cleaning up Gotham)

Memory Usage

Don't accumulate data in memory (keep the Batcomputer running smoothly)

Timeouts

Implement timeouts (even Batman needs sleep)

What's Next?

Now that you've mastered streaming, you might want to explore: