是否有更好的方法来服务和编码来自FastAPI的大型SQLAlchemy数据集？

问题描述

我想使用 FastAPI StreamingResponse返回大型数据集，并在存储库/逻辑层中，在完成查询工作之后，我将以这种方式返回数据：>

for record in query.yield_per(DATA_RECORDS_LIMIT):
            yield record.to_entity()

最初，我遇到了一些编码问题（缺少编码器方法，无法序列化日期时间等...），感谢这个https://github.com/encode/starlette/issues/419#issuecomment-470077657和这个https://fastapi.tiangolo.com/tutorial/encoder/#using-the-jsonable_encoder，我最终在Web处理程序：

...
results = get_data()

def _encoded_results():
        yield "["
        for idx,item in enumerate(results):
            if idx > 0:
                yield ","
            yield json.dumps(jsonable_encoder(item.dict()))
        yield "]"

return StreamingResponse(_encoded_results())

现在...在您问之前：是的，它有效，但是我想知道所有这一切是否必要，或者是否有更好的方法 。要添加更多上下文，第一个代码段中的记录是一个 sqlAlchemy 模型实例，然后.to_entity()将其转换为一个 Pydantic 数据实例。在第二个代码段中，我在pydantic类上调用.dict()，因此我得到了一个Python dict，可以在被jsonable_encoder之前通过.json.dumps(...)传递。

我很确定我不是唯一尝试使用FastAPI返回/流式传输非常大的数据集的人，所以我想知道是否有内置的或更好的方法来执行此操作。谢谢

注意：我的主要担心（以防万一，我不清楚）是关于给定 Pydantic 实体的事实，我需要先致电{{1} }方法，则需要先通过.dict()，最后要通过jsonable_encoder。我希望此转换在FastAPI内的某个地方实现，并在Web处理程序中隐藏。不需要TL / DR json.dumps。

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）

api api fastapi json pydantic sqlalchemy