我的系统的设计由两部分组成:
KeyValueStore
核心逻辑可以引发异常。我不知道应该如何在 FastAPI 层处理这些。
假设核心逻辑就像存储键值对的数据结构一样简单:
class KeyValueStore():
def __init__(self):
self.data = {}
def put(self, key, value):
if key in self.data:
raise RuntimeError(f'duplicate key {key}')
self.data[key] = value
def get(self, key):
# doesn't really matter what the implementation is
如您所见,如果使用之前使用过的键调用
KeyValueStore.put()
,则会导致出现时间 RuntimeError
错误。
这是在 FastAPI 层捕获的。代码看起来像这样:
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi import status
from pydantic import BaseModel
class FastAPI_KeyValuePair(BaseModel):
key: str
value: str
class FastAPI_ReturnStatus(BaseModel):
status: str
message: str|None = None
@app.post('/put_key_value')
def put_key_value(key_value_pair: FastAPI_KeyValuePair):
try:
key_value_store.put(
key_value_pair.key,
key_value_pair.value,
)
return FastAPI_ReturnStatus(status='success', message=None)
except RuntimeError as error:
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content=FastAPI_ReturnStatus(
status='error',
message=str(error),
)
)
在这种特殊情况下,我想返回 409 冲突状态代码,但我不确定如何在返回一些 JSON 正文的同时执行此操作。
正常的返回语句看起来像这样:
return FastAPI_ReturnStatus(status='success', message=None)
如果我理解正确的话,FastAPI 会使用
json.dumps
将此类型序列化为 JSON。此序列化内容将作为响应正文返回。此外,状态代码自动设置为 200。
我想做的就是将此状态代码设置为其他内容,例如 409。除了使用
JSONResponse
: 之外,我找不到其他方法来做到这一点
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content=FastAPI_ReturnStatus(
status='error',
message=str(error),
)
)
我不知道这是否真的是正确的方法。我只是选择使用
JSONResponse
,因为似乎没有任何其他方法可以返回 JSON 主体,并设置错误状态代码(在本例中为 409)。
这段代码实际发生的是我得到一个异常:
TypeError: Object of type FastAPI_ReturnStatus is not JSON serializable
这让我很困惑,因为这种类型是 JSON 可序列化的。一定是这样,因为这行:
return FastAPI_ReturnStatus(status='success', message=None)
有效。
为什么
FastAPI_ReturnStatus
在此上下文中不可序列化,但如果返回类型为 FastAPI_ReturnStatus
(未嵌套在其他对象中)则可序列化?
首先,我强烈建议您查看这个答案和这个答案,这将帮助您了解 FastAPI 如何处理响应(以及幕后的序列化),从而了解错误的本质。
至于返回包含 Pydantic
BaseModel
和自定义 status_code
的 JSON 响应(再次,请查看上面的链接答案以了解更多详细信息),可以在以下选项之间进行选择。
Response
对象更改 status_code
。
from fastapi import FastAPI, Response, status
from pydantic import BaseModel
class MyModel(BaseModel):
msg: str
app = FastAPI()
@app.get('/')
async def main(response: Response):
response.status_code = status.HTTP_409_CONFLICT # or simply = 409
return MyModel(msg="test")
使用 FastAPI 的
JSONResponse
或其他(可能更快)JSON 编码器序列化模型后,返回 jsonable_encoder
,如此答案中所述。
from fastapi import FastAPI, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from pydantic import BaseModel
class MyModel(BaseModel):
msg: str
app = FastAPI()
@app.get('/')
async def main():
return JSONResponse(content=jsonable_encoder(MyModel(msg="test")), status_code=status.HTTP_409_CONFLICT)
或者,您可以按如下方式提出
HTTPException
。
from fastapi import FastAPI, HTTPException, status
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
class MyModel(BaseModel):
msg: str
app = FastAPI()
@app.get('/')
async def main():
raise HTTPException(detail=jsonable_encoder(MyModel(msg="test")), status_code=status.HTTP_409_CONFLICT)
model_dump_json()
(从 Pydantic V1 替换
json()
)将 Pydantic 模型实例转换为 JSON 编码的字符串,然后返回自定义 Response
对象,设置 status_code
和 media_type
根据需要。
from fastapi import FastAPI, Response, status
from pydantic import BaseModel
class MyModel(BaseModel):
msg: str
app = FastAPI()
@app.get('/')
async def main():
m = MyModel(msg="test")
return Response(content=m.model_dump_json(), status_code=status.HTTP_409_CONFLICT, media_type='application/json')
还可以创建自定义
Response
类,如这个答案中所示。
您可以通过在控制器功能的参数列表中包含默认响应来设置自定义状态代码:
from fastapi import Response
@app.post('/put_key_value')
def put_key_value(key_value_pair: FastAPI_KeyValuePair, response: Response):
try:
key_value_store.put(
key_value_pair.key,
key_value_pair.value,
)
return FastAPI_ReturnStatus(status='success', message=None)
except RuntimeError as error:
response.status_code = 409
return FastAPI_ReturnStatus(
status='error',
message=str(error),
)
由于这是将使用的响应对象,除非您自己返回自定义
Response
对象,因此如果发生错误,状态代码现在将为 409
而不是默认的 200
。