是否有更好的方法从 FastAPI 提供大型 SQLAlchemy 数据集并对其进行编码？

Question

我想使用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我最终在网络处理程序中得到了这个最终代码：

...
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数据实例。在第二个片段中，我调用.dict()pydantic 类，这样我就得到了一个可以在 之前dict传递的Python 。jsonable_encoder.json.dumps(...)

我很确定我不是唯一一个尝试使用 FastAPI 返回/流式传输非常大的数据集的人，所以我想知道是否有任何内置的或更好的方法来做到这一点。谢谢

注意：我主要关心的是（如果我不清楚的话），给定一个Pydantic实体，我需要首先调用该.dict()方法，然后它需要通过jsonable_encoder，最后通过json.dumps。我希望这个转换是在 FastAPI 内部的某个地方实现的，并且对 Web 处理程序隐藏。_encoded_results不需要TL/DR 。