Coming From

If you're coming from File Handling, you'll find streaming provides a more efficient way to handle large files.

Streaming Responses

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.

Streaming responses are perfect for handling large datasets or real-time updates without consuming excessive memory.

Response

When the Bat-Signal needs to stream continuously through the night sky, you'll want to use a generator or iterator as the description parameter:

Server

from robyn import Response

@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 Response(
        status_code=200,
        headers={"Content-Type": "text/plain"},
        description=signal_generator()
    )

Client

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

Parameters

Name	Type	Description	Default
status_code	int	Response status code	200
headers	Dict[str, str]	Response headers	None
description	Union[str, bytes, Generator, AsyncGenerator]	Content to stream	None

Supported Types

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

Binary

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

Text

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

Numbers

# Numbers (like Batmobile telemetry)
yield str(speed).encode()

JSON

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

Server-Sent Events

For real-time updates from the Batcomputer:

Server

@app.get("/batcomputer/events")
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")
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

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:

Always handle errors gracefully in your streaming responses to prevent connection hangs.

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 app.test_client() as client:
        response = await client.get("/bat-signal")
        signals = []
        async for signal in response.content:
            signals.append(signal)
        assert len(signals) > 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:

WebSockets - For real-time bidirectional communication
Scaling - Scale your streaming applications across multiple cores