Asynchronous Event Streaming with Django Async Generator Views

import asyncio
from typing import AsyncGenerator
from django.http import StreamingHttpResponse
from django.views import View
 
class ServerSentEventsView(View):
    async def event_stream(self) -> AsyncGenerator[str, None]:
        try:
            for i in range(10):
                await asyncio.sleep(1)
                yield f"data: {{\"event_id\": {i}, \"status\": \"processing\"}}\n\n"
            yield "data: {\"status\": \"complete\"}\n\n"
        except asyncio.CancelledError:
            # Handle client disconnect gracefully
            pass
 
    async def get(self, request, *args, **kwargs):
        response = StreamingHttpResponse(
            self.event_stream(),
            content_type="text/event-stream"
        )
        response["Cache-Control"] = "no-cache"
        response["X-Accel-Buffering"] = "no"
        return response