如何从FastAPIendpoint（Python）返回生成器/映射对象？

Question

我正在开发我的第一个FastAPI和Pydantic项目，在尝试从endpoint返回生成器时遇到了一个问题。问题在下面，任何投入都将不胜感激！

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

我的伪代码:

@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

这样，我在运行endpoint时出错

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中，我可以看到endpoint被执行到

{ key_1: key_2 }

很奇怪，我花了相当长的时间调试，但无法整理出来。如何修复上面提到的ValueError？非常感谢您的预先投入！

Answer 1

< li >当格式为异步时，您应该等待它或只删除异步 < li>Swagger (OpenAPI)告诉您它将返回什么，因为response _ model = Iterable[record model]它并不意味着这就是您将实际返回的内容 < li >您可以使用列表理解，而不是< code >映射 < li >在响应模型< code>RecordModel中，您可以从删除它开始。然后API会返回。 < ul > < li >但是，这样你会大摇大摆地丢失一些文档 < li >另一个解决方案是检查(并修正)模型是否真正符合您返回的内容。

@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

经过一段时间的修改，虽然我仍然没有解决上面文章中提到的＜code＞ValueError＜/code＞，但我找到了一个解决方法——在处理数据库大查询时，使用分页查询而不是返回生成器作为查询结果。

如此改进和有效的伪代码：

@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用户。