flask-smorest流式响应返回空JSON及断言错误问题求助

问题根源 #

1. 空JSON列表问题：@blp.response(200, CurrencySchema(many=True)) 装饰器无法正确处理生成器返回值。该装饰器会将生成器转为列表，再用many=True的Schema序列化每个元素——但你生成器yield的是字典列表，每个列表会被当成单个Currency对象解析，而列表不符合Schema结构，最终输出空JSON对象。
2. AssertionError问题：直接返回Response(generate_rates())时，生成器产出的是Python字典列表，而非Flask要求的字节/字符串，触发断言错误。

修复代码 #

方案1：流式输出标准JSON数组（与原响应格式兼容） #

移除@blp.response装饰器，手动处理JSON序列化和流式输出：

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

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):
    def get(self):
        currency = request.args.get('currency')        
        results = asyncio.run(func_that_fetches_currency_rates_from_three_APIs(currency))
        schema = CurrencySchema()

        def generate_rates():
            batch_size = 100
            # 输出JSON数组开头
            yield b'['
            first_batch = True
            for i in range(0, len(results), batch_size):
                batch = results[i:i+batch_size]
                # 序列化批次内的每个元素
                serialized_items = [schema.dump(item) for item in batch]
                # 转为JSON字符串并编码为字节
                batch_json = json.dumps(serialized_items).encode('utf-8')
                # 非首个批次前添加逗号分隔
                if not first_batch:
                    yield b','
                else:
                    first_batch = False
                yield batch_json
            # 输出JSON数组结尾
            yield b']'
        
        return Response(generate_rates(), mimetype='application/json')

方案2：JSON Lines格式（更适合流式场景） #

如果前端支持JSON Lines（每行一个JSON对象），可以用更简洁的实现：

@blp.route("/test")
class Test(MethodView):
    def get(self):
        currency = request.args.get('currency')        
        results = asyncio.run(func_that_fetches_currency_rates_from_three_APIs(currency))
        schema = CurrencySchema()

        def generate_rates():
            batch_size = 100
            for i in range(0, len(results), batch_size):
                batch = results[i:i+batch_size]
                for item in batch:
                    # 序列化单个对象，添加换行符后编码为字节
                    item_json = json.dumps(schema.dump(item)) + '\n'
                    yield item_json.encode('utf-8')
        
        return Response(generate_rates(), mimetype='application/jsonlines')

关键说明 #

• 必须移除@blp.response装饰器：该装饰器是为常规非流式响应设计的，会干扰流式输出的序列化逻辑。
• 手动序列化并编码为字节：生成器必须产出字节流，因此需要用json.dumps()将Python对象转为JSON字符串，再用encode('utf-8')转为字节。
• 保持格式合法性：如果输出标准JSON数组，要手动处理开头、结尾和批次间的逗号，避免生成无效JSON。

内容的提问来源于stack exchange，提问作者FahdS