Flask Stream-Helfer-API
stream_with_contextWird verwendet, um den Request-Kontext in Generatorfunktionen aufrechtzuerhalten, sodass Objekte wie request und session während Streaming-Antworten weiterhin zugänglich bleiben.
stream_with_context
| Parameter | Type | Beschreibung |
|---|---|---|
| generator_or_function | iterator / callable | Eine Generatorfunktion oder ein aufrufbares Objekt. Wird als Dekorator oder Wrapper-Funktion verwendet. |
Zwei Nutzungsmodi
| Nutzung | Beschreibung |
|---|---|
| Dekorator-Muster | @stream_with_context dekoriert die Generatorfunktion |
| Wrapper-Modus | stream_with_context(generator()) umschließt einen bereits erstellten Generator |
Hinweise: Sobald das Senden des Antwortkörpers beginnt, können die Antwort-Header nicht mehr geändert werden. Wenn der Generator auf die Session zugreifen muss, müssen Sie zuerst in der View-Funktion auf die Session zugreifen, um das Setzen des Vary: Cookie-Headers auszulösen.
Codebeispiele
Beispiel
from flask import Flask, stream_with_context, request, Response
app = Flask(__name__)
@app.get("/stream")
def streamed_response():
"""Beispiel für eine Streaming-Antwort: Daten Zeile für Zeile senden"""
@stream_with_context
def generate():
# Auch nachdem mit dem Senden der Antwort begonnen wurde, kann weiterhin auf request zugegriffen werden
yield f"data: Hallo {request.args.get('name', 'EXAMPLE')}"\n"
for i in range(5):
yield f"data: Zeile {i}"\n"
yield "data: [DONE]\n"
return Response(generate(), mimetype="text/event-stream")
@app.get("/stream-wrap")
def stream_wrap():
"""Wrapper-Modus"""
def generate():
yield "Hallo "
yield request.args.get("name", "Welt")
yield "!"
# Verwende stream_with_context, um den bereits erstellten Generator zu umschließen
return Response(stream_with_context(generate()))
app = Flask(__name__)
@app.get("/stream")
def streamed_response():
"""Beispiel für eine Streaming-Antwort: Daten Zeile für Zeile senden"""
@stream_with_context
def generate():
# Auch nachdem mit dem Senden der Antwort begonnen wurde, kann weiterhin auf request zugegriffen werden
yield f"data: Hallo {request.args.get('name', 'EXAMPLE')}"\n"
for i in range(5):
yield f"data: Zeile {i}"\n"
yield "data: [DONE]\n"
return Response(generate(), mimetype="text/event-stream")
@app.get("/stream-wrap")
def stream_wrap():
"""Wrapper-Modus"""
def generate():
yield "Hallo "
yield request.args.get("name", "Welt")
yield "!"
# Verwende stream_with_context, um den bereits erstellten Generator zu umschließen
return Response(stream_with_context(generate()))