python / intermediate
Snippet
Asynchrones Streaming von Server-Sent Events via StreamingHttpResponse
StreamingHttpResponse in Kombination mit asynchronen Generatorfunktionen ermöglicht Echtzeit-Event-Streaming, ohne Worker-Threads zu blockieren. Das Setzen des SSE-Content-Typs text/event-stream überträgt Aktualisierungen kontinuierlich an Client-Verbindungen.
snippet.py
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import asynciofrom django.http import StreamingHttpResponseasync def event_generator():for count in range(5):await asyncio.sleep(1)yield f"data: {{'step': {count}}}\n\n"async def sse_stream_view(request):response = StreamingHttpResponse(event_generator(),content_type='text/event-stream')response['Cache-Control'] = 'no-cache'return response
django
Erklärung
1
async def event_generator():
Definiert einen asynchronen Generator, der zeitlich gestaffelte Text-Payload-Events liefert.
2
yield f"data: {{'step': {count}}}\n\n"
Generiert korrekt formatierte Server-Sent-Event-Strings gemäß dem SSE-Standardformat.
3
response = StreamingHttpResponse(event_generator(), content_type='text/event-stream')
Bettet den Generator in Djangos Streaming-Response-Objekt ein, das für SSE-Streams konfiguriert ist.