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.