如何从 FastAPI 端点 (Python) return 一个 generator/map 对象？

Question

我正在处理我的第一个 FastAPI 和 Pydantic 项目，在尝试从端点 return 生成器时遇到了问题。问题如下，任何输入将不胜感激！

我有一个 API 端点，我首先从数据库中获取数据记录，然后使用 map 函数格式化每条记录。但是，当 return 将映射结果发送到端点时，键值映射出错了。请注意，为了性能（大数据量），我想将端点的 return 类型保留为生成器。

我的伪代码：

@app.get("/records", response_model=Iterable[RecordModel])
async def get_records() -> Iterable[RecordModel]:
    # {queried_records} is a generator returned from the database query
    queried_records = get_records_from_database() 
    formatted_records = map(lambda record: __format(record), 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

据此，当运行端点

时出现错误

ValueError: [ValueError('dictionary update sequence element #0 has length 3; 2 is required'), TypeError('vars() argument must have __dict__ attribute')]

如果我将 __format 方法更改为

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]
    )
    return formatted_record

从 Swagger UI，我可以看到端点被执行到

的响应主体

{ key_1: key_2 }

很奇怪，我调试了好久，也没能解决。如何解决上面提到的ValueError？非常感谢您的提前投入！

Answer 1

当格式为异步时，您应该等待它或直接删除异步
Swagger (OpenAPI) 告诉你它将 return 因为 response_model=Iterable[RecordModel] 这并不意味着那是你实际 [=31] =]
您可以使用列表理解代替 map
重新加载响应模型 RecordModel 您可以从删除它开始。然后 API 会 return 无论如何。
- 然而，您将在 Swagger 中丢失一些文档
- 另一个解决方案是检查（并修复）模型是否真的适合您 return。

@app.get("/records")
async def get_records() -> Iterable[RecordModel]: return [
    __format(record) for record 
    in get_records_from_database()
]

def __format(
    queried_record: Dict[str, Union[str, HttpUrl]
) -> Union[RecordModel, None]: return RecordModel(
    key_1 = queried_record[key_a],
    key_2 = queried_record[key_b],
    key_3 = queried_record[key_c]
)

Answer 2

修修补补一段时间后，虽然还是没有解决上面提到的ValueErrorpost，但我找到了一个work-around——在处理数据库大查询时，使用用于查询的分页，而不是返回生成器作为查询结果。

如此改进和工作的伪代码：

@app.get("/records", response_model=List[RecordModel])
async def get_records(
    offset: int = 0, # start position of the query
    limit: int = 1000 # size of the query
) -> List[RecordModel]:
    queried_records = get_records_from_database(offset, limit) 
    formatted_records = map(lambda record: __format(record), queried_records)
    return list(formatted_records)

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

至此，我将使用分页处理大型数据库查询的责任移交给 API 用户。

如何从 FastAPI 端点 (Python) return 一个 generator/map 对象？

How to return a generator/map object from FastAPI endpoint (Python)?

rest

generator

python-3.x

pydantic

fastapi