HttPException
向客户端返回 HTTP 错误响应,可以使用HTTPException。
from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"foo": "The Foo Wrestlers"}
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
return {"item": items[item_id]}
添加自定义响应头
from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"foo": "The Foo Wrestlers"}
@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
if item_id not in items:
raise HTTPException(
status_code=404,
detail="Item not found",
headers={"X-Error": "There goes my error"},
)
return {"item": items[item_id]}
自定义异常处理器
使用@app.exception_handler()可以添加自定义异常控制器
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class UnicornException(Exception): # 异常类可以写在其他py中
def __init__(self, name: str):
self.name = name
app = FastAPI()
@app.exception_handler(UnicornException) # 这块代码必须在main.py文件才行,写其他文件报错
async def unicorn_exception_handler(request: Request, exc: UnicornException):
return JSONResponse(
status_code=418,
content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
)
@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
if name == "yolo":
raise UnicornException(name=name)
return {"unicorn_name": name}
ps:经过尝试@app.exception_handler() 使用这样装饰器的方式来自定义异常的时候,代码必须写在main.py文件中,不然报错。
覆盖默认异常
RequestValidationError请求验证异常
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
item_name: str
item_id: int
@app.post("/items/{item_id}", response_model=Item)
async def read_item(item: Item):
return item
默认的验证参数提示又臭又长,如果想要自己定义提示,就需要覆盖RequestValidationError
from fastapi import FastAPI, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
item_name: str
item_id: int
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc: RequestValidationError):
message = ""
for error in exc.errors():
message += ".".join(error.get("loc")) + ":" + error.get("msg") + ";"
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={
"code": 1,
"message": message,
"data": None
}
)
@app.post("/items/{item_id}", response_model=Item)
async def read_item(item: Item):
return item
添加全局异常校验方式
使用装饰器
第一种就是上面通过@app.exception_handler()装饰器的方法来添加
非装饰器
1.使用FastAPI实例的add_exception_handler()方法
exc_class_or_status_code:状态码,或错误类
from fastapi import FastAPI, status, HTTPException, Request
from fastapi.responses import JSONResponse
app = FastAPI()
class TokenException(Exception):
pass
async def token_exception_handler(request, exc):
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={
"code": 2,
"message": "登录失效,请重新登录",
"data": None
}
)
# app.add_exception_handler(TokenException, token_exception_handler)
app.add_exception_handler(401, token_exception_handler)
@app.post("/items/{item_id}")
async def read_item(request: Request):
authorization: str = request.headers.get("Authorization")
if not authorization:
# raise TokenException()
raise HTTPException(status_code=401)
app.add_exception_handler(TokenException, token_exception_handler)和
raise TokenException()配合使用
2.创建fastApi实例时,传入exception_handlers参数并传入一个字典
from fastapi import FastAPI, status, HTTPException, Request
from fastapi.responses import JSONResponse
class TokenException(Exception):
pass
async def token_exception_handler(request, exc):
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={
"code": 2,
"message": "登录失效,请重新登录",
"data": None
}
)
exception_handlers_dict = {
401: token_exception_handler,
TokenException: token_exception_handler
}
app = FastAPI(exception_handlers=exception_handlers_dict)
@app.post("/items/{item_id}")
async def read_item(request: Request):
authorization: str = request.headers.get("Authorization")
if not authorization:
# raise TokenException()
raise HTTPException(status_code=401)