Snippet

Asynchronous StreamingHttpResponse using Generator Exception Propagation

Django supports async generator streaming responses. Utilizing asyncio.CancelledError inside an async generator allows clean resource reclamation when HTTP clients sever connections prematurely.

snippet.py

python

import asyncio
from typing import AsyncGenerator
from django.http import StreamingHttpResponse
 
async def generate_telemetry_stream() -> AsyncGenerator[str, None]:
    try:
        for step in range(5):
            await asyncio.sleep(0.1)
            yield f"event: metric\ndata: {{\"step\": {step}}}\n\n"
    except asyncio.CancelledError:
        # Clean up socket resources on client disconnect
        pass
 
async def telemetry_stream_view(request):
    return StreamingHttpResponse(
        generate_telemetry_stream(),
        content_type="text/event-stream"
    )

django

Breakdown

async def generate_telemetry_stream() -> AsyncGenerator[str, None]:

Declares an asynchronous generator yielding data chunks without blocking the main event loop.

yield f"event: metric\ndata: {{\"step\": {step}}}\n\n"

Emits Server-Sent Event formatted string data to the downstream response stream.

except asyncio.CancelledError:

Catches cancellation signals triggered when client connection drops during streaming.

Previous snippet Next snippet

More Expert Python exercises →

Asynchronous StreamingHttpResponse using Generator Exception Propagation

Related