1.异常处理
从官方文档的理解
有两个主要异常处理程序，1. HTTPException处理程序 2.RequestValidationError处理程序
from starlette.exceptions import HTTPException as StarletteHTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.exception_handlers import request_validation_exception_handler
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder
# 错误处理 全局设置
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
print('异常发生')
return JSONResponse(staus_code=错误代码xxx, content=jsonable_encoder({
'message': f"Oops! {exc.errors()} did something. There goes a rainbow..."
}))
# 请求校验异常
@app.exception_handler(RequestValidationError)
async def validation_error_handler(request, exc):
print('请求校验异常')
return JSONResponse(staus_code=错误代码xxx, content=jsonable_encoder({
'message': f"Oops! {exc.errors()} did something. There goes a rainbow..."
}))
# 重用http异常处理程序
@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):
print('一个http异常')
return await http_exception_handler(request, exc)
# 重用请求校验异常处理程序
@app.exception_handler(RequestValidationError)
async def custom_validation_error_handler(request, exc):
print('一个请求校验异常')
return await request_validation_exception_handler(request, exc)
# 自定义异常处理器
class CustomException(Exception):
def __init__(self, name: str)
self.name = name
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
return JSONResponse(status_code=404, content={xxx})
2.put和patch
使用put方法时，如果没有传入全部参数，且空出的参数是有默认值的，会被填充默认值
可以使用.dict(exclude_unset=True)来忽略默认值
或者使用储存模型(对象)的副本，使用 copy(update=update_data)来更新属性
部分更新的流程：
- 检索储存的数据
- 将数据放入pydnatic模型中
- dict从输入模型生成一个没有默认值的（使用exclude_unset)
- 创建储存模型的副本，使用接受到的部分更新（使用update参数）更新其属性
- 将复制的模型装换为可以储存在数据库中的模型（如：jsonable_encoder)
- 将数据保存到数据库
- 返回更新的模型
使用部分更新 需要考虑接受到可以忽略所有属性的部分更新，需要有一个模型，标记全部属性都可选(设置默认值或者None) 可以派生一个子类，标记为可选，，
@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
stored_item_data = items[item_id]
stored_item_model = Item(**stored_item_data)
update_data = item.dict(exclude_unset=True)
updated_item = stored_item_model.copy(update=update_data)
items[item_id] = jsonable_encoder(updated_item)
return updated_item
3.依赖注入
使用Depends
何时有需要：
- 有共享逻辑
- 共享数据库连接
- 强制执行安全性，身份认证，角色要求
- 。。。
# 依赖项 要求是可调用的
async def common_parmeters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
return {'q': q, 'skip': skip, 'limit': limit}
class CommonQueryParams():
def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
self.test = 'test'
@app.get('/dp')
async def read_users(commons: dict = Depends(common_parmeters)):
return commons
# commons: dict = Depends(common_parmeters)
# 如果是一个类对象，可以这么写
commons: common_parmeters = Depends(common_parmeters)
或者
commons: common_parmeters = Depends()
1. 依赖缓存
如果针对同一路径操作多次声明一个依赖项，如：多个依赖项具有一个公共的子依赖项，则fastapi会只调用一次子依赖项，并且将返回的值保存在“缓存”中， 将其传给该特定请求中需要它的所有“依赖项”，而不是针对一个请求多次调用依赖项。
如果有需求需要在同一请求中的每个步骤（可能多次）调用依赖项，而不使用cache值，可以使用user_cache=False
async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
return {}
2.对于不需要返回值的依赖
可以添加dependencies到路径操作装饰器中 ,它需要传入一个list的Depends()
在依赖项中也可以抛出异常，也可以有返回值，但是将不会被使用
@app.get('/items', dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
return [{}]
3.全局依赖
对于某些需应用到所有接口的依赖项，可以这么做：
app = FastAPI(dependencies=[Depends(verify_token), Dependens(verify_key)])
4. 依赖的yield
FastAPI支持依赖项在完成后再执行一些额外的步骤
需要使用python3.7或者更高版本，或者再python3.6中安装 backports
pip install async-exit-stack async-generator
例：
async def get_db():
db = DBSession() ->2
try:
yield db ->3
finally:
db.close() -> 6
@app.get('/db')
async def db_op(db: Session = Depends(get_db)): -> 1
-> 4
return 'ok' -> 5
# 访问db接口时，会调用依赖项，获取db，期间函数可以操作db对象，直到返回，返回后，get_db会执行finally的db.close()
