Quiero devolver un gran conjunto de datos usando FastAPI StreamingResponse y en la capa de repositorio/lógica, después de hacer mis consultas, devuelvo los datos de esta manera:

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

Inicialmente tuve algunos problemas de codificación (faltaba el método del codificador, no podía serializar una fecha y hora, etc.) y gracias a esto https://github.com/encode/starlette/issues/419#issuecomment-470077657 y este https: //fastapi.tiangolo.com/tutorial/encoder/#using-the-jsonable_encoder Terminé con este código final en el controlador 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())

Ahora... antes de que preguntes: sí, funciona , pero me preguntaba si todo esto es necesario o si hay una mejor manera de hacerlo . Para agregar más contexto, el registro en el primer fragmento es una instancia del modelo SQLAlchemy y .to_entity() lo transforma en una instancia de datos de Pydantic . En el segundo fragmento, llamo a .dict() en la clase pydantic, por lo que obtengo un jsonable_encoder dict de ser .json.dumps(...) .

Estoy bastante seguro de que no soy el único que intenta usar FastAPI para devolver/transmitir un conjunto de datos muy grande, por lo que me pregunto si hay algo integrado o una mejor manera de hacerlo. Gracias

Nota: mi principal preocupación (en caso de que no haya quedado claro) se trata del hecho de que dada una entidad Pydantic , primero necesito llamar al método .dict() , luego debe hacerlo a través de jsonable_encoder y finalmente a través json.dumps . Desearía que esta transformación se implementara en algún lugar dentro de FastAPI y se ocultara del controlador web. TL/DR _encoded_results no debería ser necesario.