API вспомогательных функций потоковой передачи Flask
stream_with_contextИспользуется для поддержания контекста запроса в функциях-генераторах, чтобы такие объекты, как request и session, оставались доступными во время потоковых ответов.
stream_with_context
| Параметр | Type | Описание |
|---|---|---|
| generator_or_function | итератор / вызываемый объект | Функция-генератор или вызываемый объект. Используется как декоратор или функция-обёртка. |
Два режима использования
| Использование | Описание |
|---|---|
| Паттерн «Декоратор» | @stream_with_context декорирует функцию-генератор |
| Режим обёртки | stream_with_context(generator()) оборачивает уже созданный генератор |
Примечания: Как только тело ответа начинает отправляться, заголовки ответа больше нельзя изменять. Если генератору нужен доступ к сессии, необходимо сначала обратиться к сессии в функции представления, чтобы вызвать установку заголовка Vary: Cookie.
Примеры кода
Пример
from flask import Flask, stream_with_context, request, Response
app = Flask(__name__)
@app.get("/stream")
def streamed_response():
"""Пример потокового ответа: отправка данных строка за строкой"""
@stream_with_context
def generate():
# Даже после того, как ответ начал отправляться, доступ к request всё ещё возможен
yield f"data: Hello {request.args.get('name', 'EXAMPLE')}\n"
for i in range(5):
yield f"data: Line {i}"\n"
yield "data: [DONE]\n"
return Response(generate(), mimetype="text/event-stream")
@app.get("/stream-wrap")
def stream_wrap():
"""Режим обёртки"""
def generate():
yield "Hello "
yield request.args.get("name", "World")
yield "!"
# Используйте stream_with_context, чтобы обернуть уже созданный генератор
return Response(stream_with_context(generate()))
app = Flask(__name__)
@app.get("/stream")
def streamed_response():
"""Пример потокового ответа: отправка данных строка за строкой"""
@stream_with_context
def generate():
# Даже после того, как ответ начал отправляться, доступ к request всё ещё возможен
yield f"data: Hello {request.args.get('name', 'EXAMPLE')}\n"
for i in range(5):
yield f"data: Line {i}"\n"
yield "data: [DONE]\n"
return Response(generate(), mimetype="text/event-stream")
@app.get("/stream-wrap")
def stream_wrap():
"""Режим обёртки"""
def generate():
yield "Hello "
yield request.args.get("name", "World")
yield "!"
# Используйте stream_with_context, чтобы обернуть уже созданный генератор
return Response(stream_with_context(generate()))