如何在FastAPI端点返回generator/map对象并解决ValueError报错

问题根源 #

• 异步函数误用：你定义的__format为异步函数，但内置同步函数map不会等待协程执行，生成器内的元素为未执行的协程对象，而非预期的RecordModel实例，FastAPI序列化协程对象时触发类型错误，你观察到的异常返回值就是序列化协程的错误结果。
• 响应模型声明不规范：Iterable[RecordModel]不能直接作为FastAPI的response_model参数值，列表响应需声明为list[RecordModel]，FastAPI支持自动迭代生成器完成序列化，不会全量加载数据到内存。

修复方案 #

方案1：同步格式化（推荐，适合无异步IO的格式化逻辑） #

你的__format函数没有异步IO操作，无需定义为异步函数，直接去掉async修饰符即可解决核心问题，代码示例：

@app.get("/records", response_model=list[RecordModel])
async def get_records() -> Iterable[RecordModel]:
    # queried_records 是数据库查询返回的generator
    queried_records = get_records_from_database() 
    # 直接传入同步格式化函数，map返回的生成器可被FastAPI直接处理
    formatted_records = map(__format, queried_records)
    return formatted_records

# 去掉async修饰符，改为同步函数
def __format(queried_record: Dict[str, Union[str, HttpUrl]]) -> Union[RecordModel, None]:
    formatted_record = RecordModel(
        key_1 = queried_record[key_a],
        key_2 = queried_record[key_b],
        key_3 = queried_record[key_c]
    )
    return formatted_record

该方案仍保留生成器的低内存特性，符合大数据量场景的性能要求。

方案2：异步流式输出（适合格式化逻辑包含异步IO的场景） #

如果你的格式化逻辑确实需要异步IO调用，需使用异步生成器配合StreamingResponse实现流式输出：

from fastapi.responses import StreamingResponse
import json

async def __format(queried_record: Dict[str, Union[str, HttpUrl]]) -> Union[RecordModel, None]:
    # 此处可添加异步IO操作
    formatted_record = RecordModel(
        key_1 = queried_record[key_a],
        key_2 = queried_record[key_b],
        key_3 = queried_record[key_c]
    )
    return formatted_record

async def stream_records():
    queried_records = get_records_from_database()
    yield "["
    first_record = True
    for record in queried_records:
        if not first_record:
            yield ","
        first_record = False
        formatted_record = await __format(record)
        yield formatted_record.model_dump_json()
    yield "]"

@app.get("/records")
async def get_records():
    return StreamingResponse(stream_records(), media_type="application/json")

内容的提问来源于stack exchange，提问作者Mao J.Y. Yu