在FastAPI框架中,Depends
是一个关键的功能,它允许开发者通过依赖注入来管理和重用代码。这在处理API的分页和过滤逻辑时尤其有用,因为它可以将这些逻辑抽象化,从而减少冗余代码并提高效率。
通过Depends
,我们可以定义一个函数,该函数负责获取和验证分页参数(如skip
和limit
)以及过滤参数(如基于名称或价格的筛选)。
在路由函数中,我们通过Depends
将这些参数作为依赖项注入,FastAPI会在调用路由函数之前自动执行这些依赖函数,并将结果作为参数传递给路由函数。这种方式不仅使代码更加模块化,而且使得参数处理逻辑可以在多个路由中重用,从而简化了API的开发和维护。
在下段 Python 代码中,commons
是一个依赖函数 common_parameters
的结果,它通过 FastAPI 的依赖注入系统被传递给路由处理函数。这里的 Annotated
来自 typing
模块,它用于添加额外的类型信息或元数据到已有的类型上,但在 FastAPI 中,它主要用于添加额外的文档信息到参数上。
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return commons
@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
return commons
让我们逐步分析 commons
函数:
-
common_parameters
是一个异步函数,它接受三个参数:q
: 一个可选的字符串,用于搜索或过滤。skip
: 一个整数,默认值为 0,用于分页,表示要跳过多少条记录。limit
: 一个整数,默认值为 100,用于分页,表示每页显示多少条记录。
-
common_parameters
函数返回一个字典,包含它接收到的参数。 -
Annotated[dict, Depends(common_parameters)]
是一个注解,它告诉 FastAPIcommons
参数应该通过调用common_parameters
函数来获取。Depends
是 FastAPI 用于声明依赖的装饰器。 -
在路由处理函数
read_items
和read_users
中,commons
参数被用作一个依赖项。这意味着在调用这些函数之前,FastAPI 会自动调用common_parameters
函数,并将其返回的字典作为commons
参数的值。 -
read_items
和read_users
函数简单地返回它们接收到的commons
字典,这个字典包含了查询参数和分页信息。
在实际应用中,common_parameters
函数可以用来处理来自客户端的通用查询参数
下面是一个简化后的代码示例,它展示了如何在 FastAPI 中使用依赖Depends
注入来处理分页和过滤逻辑:
from fastapi import FastAPI, Depends, HTTPException, Query
from typing import List, Optional
from pydantic import BaseModel
from sqlalchemy.orm import Session
# 假设我们有一个数据库会话依赖项
def get_db() -> Session:
# 这里应该是创建或获取数据库会话的逻辑
...
# 定义分页参数的 Pydantic 模型
class Pagination(BaseModel):
skip: int = Query(default=0, ge=0, description="Number of items to skip")
limit: int = Query(default=10, gt=0, description="Maximum number of items to return")
# 定义过滤参数的 Pydantic 模型
class Filter(BaseModel):
name: Optional[str] = Query(None, description="Filter by name")
price: Optional[float] = Query(None, description="Filter by price")
# 假设我们有一个模型类
class Item(BaseModel):
id: int
name: str
description: str = None
price: float
tax: float = None
# 路由处理函数,使用分页和过滤依赖项
@app.get("/items/", response_model=List[Item])
async def read_items(
db: Session = Depends(get_db),
pagination: Pagination = Depends(),
filter: Filter = Depends()
):
# 应用过滤逻辑
query = db.query(Item)
if filter.name:
query = query.filter(Item.name == filter.name)
if filter.price:
query = query.filter(Item.price <= filter.price) # 使用 <= 以便包含指定价格
# 应用分页逻辑
items = query.offset(pagination.skip).limit(pagination.limit).all()
if not items:
raise HTTPException(status_code=404, detail="Items not found")
return items
通过 curl 发送一个请求体,可以使用以下命令:
curl -X POST "http://localhost:8000/items/" -H "Content-Type: application/json" -d '{"skip": 0, "limit": 5, "name": "example"}'
在这个 curl 命令中:
- -X POST 指定了请求方法为 POST。
- -H “Content-Type: application/json” 设置了请求头,告诉服务器我们发送的数据是 JSON 格式。
- -d ‘{“skip”: 0, “limit”: 5, “name”: “example”}’ 是我们要发送的 JSON 格式的请求体数据。
这样,你就可以通过 curl 命令向 FastAPI 应用发送包含分页和过滤参数的请求体了。
在这个简化的版本中:
- 我们直接在
Pagination
和Filter
类中使用了Query
来定义默认值和从查询参数中获取值。 read_items
函数现在直接依赖于Pagination
和Filter
类的实例,这些实例会自动从请求的查询参数中获取值。- 我们移除了单独的
pagination_parameters
和filter_parameters
依赖函数,因为Pagination
和Filter
类已经包含了所需的逻辑。 Query
函数用于直接从查询参数中获取值,并可以设置默认值和其他验证条件。
这个简化的代码仍然实现了分页和过滤逻辑,同时减少了代码的复杂性。