当前位置：首页 > article >正文

FastAPI 进阶：使用 Pydantic 验证器增强 Query 参数验证

article 2024/11/16 15:57:17

在 FastAPI 中，为 Query 类参数添加更复杂的验证逻辑可以通过以下几种方法实现：

使用 Pydantic 验证器：
Pydantic 允许你在模型中定义自定义验证器。这些验证器可以用于 Query 参数，以实现复杂的验证逻辑。

from fastapi import FastAPI, Query, Depends, HTTPException
from pydantic import BaseModel, validator

app = FastAPI()

class QueryParameters(BaseModel):
    q: str

    @validator('q')
    def check_custom_logic(cls, value):
        # 这里可以添加任何复杂的逻辑
        if 'forbidden' in value:
            raise ValueError('The value is not allowed')
        return value

@app.get("/items/")
async def read_items(q: QueryParameters = Depends()):
    return {"q": q.q}

使用前置依赖函数：
通过在路由函数中使用 Depends，你可以创建一个前置依赖函数来执行复杂的验证逻辑。

from fastapi import FastAPI, Depends, Query, HTTPException

app = FastAPI()

def complex_query_validator(query: str):
    if not query.startswith("allowed_"):
        raise HTTPException(status_code=400, detail="Query parameter does not follow the rules")
    return query

@app.get("/items/")
async def read_items(q: str = Depends(complex_query_validator)):
    return {"q": q}

使用正则表达式：
Query 类支持 regex 参数，你可以使用它来定义一个正则表达式，用于验证查询参数的格式。

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(q: str = Query(..., regex="^[a-zA-Z0-9]+$")):
    return {"q": q}

组合使用 Query 参数：
你可以组合使用多个 Query 参数，并在路由函数中实现逻辑来验证这些参数的组合。

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(min_price: float = Query(gt=0), max_price: float = Query(gt=0)):
    if min_price >= max_price:
        raise ValueError("min_price should be less than max_price")
    return {"min_price": min_price, "max_price": max_price}

使用自定义依赖项类：
你可以创建自定义的依赖项类，这些类继承自 Depends 并实现 __call__ 方法，以封装复杂的验证逻辑。

from fastapi import FastAPI, Depends, Query

app = FastAPI()

class CustomQuery:
    def __call__(self, query: str):
        if not query.isalpha():
            raise ValueError("Query must contain only letters")
        return query

@app.get("/items/")
async def read_items(q: str = Depends(CustomQuery())):
    return {"q": q}

使用外部库：
如果需要，你可以使用外部验证库（如 Marshmallow 或 Cerberus）来实现复杂的验证逻辑，并将验证结果作为依赖项注入路由函数。

通过这些方法，你可以为 FastAPI 应用中的 Query 类参数添加几乎任何复杂度的验证逻辑，确保输入数据的准确性和安全性。

在 FastAPI 应用中，为 Query 参数编写 Pydantic 验证器可以让你实现复杂的验证逻辑。以下是如何为 Query 参数创建和使用 Pydantic 验证器的步骤：

定义 Pydantic 模型：
创建一个 Pydantic 模型，并在模型中定义你的 Query 参数以及相应的验证器。
编写验证器：
使用模型中的 @validator 装饰器来创建一个验证器函数，该函数将包含你的验证逻辑。
使用依赖注入：
在路由函数中，使用 Depends 来注入你的模型实例，这样 FastAPI 就会在调用路由函数之前执行验证逻辑。

下面是一个示例代码，展示了如何为 Query 参数编写 Pydantic 验证器：

from fastapi import FastAPI, Depends, Query
from pydantic import BaseModel, Field, validator
from typing import Union, Optional

app = FastAPI()

# 定义 Pydantic 模型
class QueryParameters(BaseModel):
    q: Optional[str] = None
    size: Optional[int] = Query(default=None, gt=0)

    # 验证器，确保 'q' 参数不为空，且 'size' 大于 0
    @validator('q', pre=True, always=True)
    def check_q_not_empty(cls, v):
        if v is None or v == "":
            raise ValueError("Query parameter 'q' cannot be empty")
        return v

    @validator('size', pre=True, allow_reuse=True)
    def check_size_positive(cls, v):
        if v is not None and v <= 0:
            raise ValueError("Query parameter 'size' must be greater than 0")
        return v

# 依赖项函数，返回验证后的 QueryParameters 实例
def validate_query_params(params: QueryParameters = Depends()):
    return params

# 路由函数
@app.get("/items/")
async def read_items(query_params: QueryParameters = Depends(validate_query_params)):
    results = {"q": query_params.q, "size": query_params.size}
    return results