flask-smorest 的流响应问题

Question

我在 Flask-smorest 中的流响应中遇到问题。我遵循此处的指导 - https://flask.palletsprojects.com/en/2.3.x/patterns/streaming/ 用于从我的flask-smorest应用程序流式传输响应。下面是我的代码的 MRE 版本。假设我的应用程序正在获取最终用户请求的任何货币过去 1000 天的外汇汇率。

这是没有使用流媒体的版本。它工作完美并返回 json 响应列表：

from flask import request, Response
from flask.views import MethodView
from flask_smorest import Blueprint, abort
from marshmallow import Schema, fields
import asyncio

class CurrencySchema(Schema):
    name = fields.Str()
    rate = fields.Str()
    date = fields.Str()
    source = fields.Str()

blp = Blueprint("test",__name__, description="test")

@blp.route("/test")
class Test(MethodView):
    @blp.response(200, CurrencySchema(many=True))
    def get(self):
        currency = request.args.get('currency')
        results = asyncio.run(func_that_fetches_currency_rates_from_three_APIs(
            currency))  # returns a list of dictionaries
        return results

当我运行它时，它成功运行并在我的浏览器上返回 json 响应列表，例如：

[{'name': 'USD', 'rate': '1.2333', 'date': 'Mar 21, 2024', 'source': 'currency.com'}, 
 {'name': 'USD', 'rate': '1.2121', 'date': 'Mar 22, 2024', 'source': 'currency.com'}, 
 .................so on and so forth up to 1000 jsons]

现在，是我尝试流式传输响应的部分。我对我的代码进行了以下更改：

@blp.route("/test")
class Test(MethodView):
    @blp.response(200, CurrencySchema(many=True))
    def get(self):
        currency = request.args.get('currency')        
        results = asyncio.run(func_that_fetches_currency_rates_from_three_APIs(
            currency))  # returns a list of dictionaries
        def generate_rates():
             batch_size = 100
             for i in range(0, len(results), batch_size):
                  yield results[i:i+batch_size]
        return generate_rates()

这奇怪地返回了 50 个空 json 响应的列表：

[{}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {},
 {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {},
 {}, {}, {}, {}, {}, {}, {}, {}]

我也尝试过这个，但结果相同，即空json响应列表，但另外flask-smorest给了我：

断言错误：应用程序必须写入字节

似乎 werkzeugserving.py 文件引发了问题。

@blp.route("/test")
class Test(MethodView):
    @blp.response(200, CurrencySchema(many=True))
    def get(self):
        currency = request.args.get('currency')        
        results = asyncio.run(func_that_fetches_currency_rates_from_three_APIs(
            currency))  # returns a list of dictionaries
        def generate_rates():
             batch_size = 100
             for i in range(0, len(results), batch_size):
                  yield results[i:i+batch_size]
        return Response(generate_rates(), mimetype = 'application/json')

我的整个应用程序已准备就绪，这是出现问题的最后一点。我想流式传输响应，但 Flask-smorest 中有些东西导致了问题。非常感谢您的支持。谢谢！

Answer 1

您需要生成字符串形式的响应并将其流式传输到客户端。您可以通过修改 API 来实现：

@blp.route("/test")
class Test(MethodView):
    @blp.response(200, CurrencySchema(many=True))
    def get(self):
        currency = request.args.get('currency')        
        results = asyncio.run(func_that_fetches_currency_rates_from_three_APIs(
            currency))  # returns a list of dictionaries
        def generate_rates():
             batch_size = 100
             yield '['
             for i in range(0, len(results), batch_size):
                  result_data = []
                  for result in results[i:i+batch_size]:
                      result_data += CurrencySchema().dumps(result) + ', '
                  yield result_data
             yield ']'
        return Response(stream_with_context(generate_rates()), status=200, content_type='application/json')