1. 使用异步依赖：
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI()
async def get_query_token(token: str = Depends(lambda x: x.query_params.get("token"))):
if token is None:
raise HTTPException(status_code=400, detail="Token not provided")
return token
@app.get("/items/")
async def read_items(token: str = Depends(get_query_token)):
return {"token": token}
在这个例子中，我们定义了一个异步依赖函数 get_query_token，它接收查询参数中的 token。如果 token 不存在，它会引发一个 HTTP 异常。然后，我们将这个依赖函数作为参数传递给路径操作函数 read_items。
2. 依赖嵌套：
from fastapi import FastAPI, Depends
app = FastAPI()
async def get_query_token(token: str = Depends(lambda x: x.query_params.get("token"))):
return token
async def get_db_conn():
# 这里可以放数据库连接的逻辑
return {"db_connection": "fake_db"}
async def get_current_user(token: str = Depends(get_query_token), db_conn: dict = Depends(get_db_conn)):
return {"token": token, "db_connection": db_conn}
@app.get("/items/")
async def read_items(current_user: dict = Depends(get_current_user)):
return current_user
在这个例子中，我们定义了三个依赖函数：get_query_token、get_db_conn 和 get_current_user。get_current_user 依赖了前两个，通过 Depends 进行了嵌套。路径操作函数 read_items 依赖了 get_current_user。
3. 使用共享的依赖：
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
app = FastAPI()
fake_users_db = {
"testuser": {
"username": "testuser",
"password": "testpassword",
}
}
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=401,
detail="Invalid credentials",
headers={"WWW-Authenticate": "Bearer"},
)
user = fake_users_db.get(token)
if user is None:
raise credentials_exception
return user
@app.get("/users/me")
async def read_me(current_user: dict = Depends(get_current_user)):
return current_user
在这个例子中，我们使用 OAuth2PasswordBearer 来定义一个 OAuth2 密码授权的依赖，并将其作为参数传递给路径操作函数 read_me。这允许我们在多个路径操作中共享相同的身份验证逻辑。
这些是一些高级依赖的示例。依赖是 FastAPI 中非常强大的功能，可以用于组织和重用代码逻辑。
转载请注明出处：http://www.zyzy.cn/article/detail/7413/FastAPI