Python----FastAPI(请求体、请求参数验证_(Query方式、Path方式、Field方式))
·
在构建现代Web API时,数据验证是至关重要的一环,它能确保传入的数据符合预期格式,从而避免程序崩溃和安全漏洞。
FastAPI 借助 Pydantic 库,将请求参数和请求体的验证、类型提示和自动文档生成完美地结合在一起,极大地简化了开发工作。
一、请求体
请求体是客户端向服务器发送的数据,通常用于 POST、PUT 等请求中。在 FastAPI 中,请求体最强大的验证方式是使用 Pydantic 模型。
from fastapi import FastAPI # 从fastapi库中导入FastAPI类,用于创建Web应用
from pydantic import BaseModel # 从pydantic库中导入BaseModel,用于定义数据模型和数据验证
# 创建一个FastAPI应用实例
app = FastAPI()
# 定义一个Pydantic模型User
# 这个模型定义了用户数据的结构和类型,以及可选的默认值
class User(BaseModel):
name: str # 用户名,字符串类型,必填
age: int # 年龄,整数类型,必填
pwa: str # 密码,字符串类型,必填 (注意:实际应用中密码不应直接这样传递或存储)
sex: str = "男" # 性别,字符串类型,可选,默认值为“男”
# 定义一个GET请求的根路径路由
@app.get('/')
def index():
# 当访问根路径时,返回一个简单的字典(FastAPI会自动将其序列化为JSON)
# 注意:这里返回的是一个集合 {'hello', 'world'},FastAPI会将其转换为列表 ['hello', 'world']
return {'hello', 'world'}
# 定义一个POST请求的路由 /user
@app.post('/user')
def create_user(user: dict):
# 这个函数接收一个类型为dict的user参数作为请求体
# FastAPI会尝试将请求体解析为字典,但不会进行严格的字段验证
return user
# 定义一个POST请求的路由 /user2
@app.post('/user2')
def create_user2(user: User):
# 这个函数接收一个类型为User(我们定义的Pydantic模型)的user参数作为请求体
# FastAPI会根据User模型对请求体进行自动验证。
# 如果请求体数据不符合User模型的定义(例如缺少name或age,或类型不匹配),
# FastAPI会自动返回一个422 Unprocessable Entity错误,并提供详细的验证失败信息。
return user
# 当直接运行这个脚本时,启动Uvicorn服务器
if __name__ == '__main__':
import uvicorn # 导入uvicorn库,这是一个ASGI服务器
uvicorn.run(
app='demo05:app', # 指定要运行的FastAPI应用实例(格式为:模块名:应用变量名)
host='127.0.0.1', # 服务器监听的IP地址
port=8000, # 服务器监听的端口
reload=True # 启用热重载,当代码文件发生变化时,服务器会自动重启
)
二、请求参数验证_Query方式
查询参数位于URL的?后面,如 http://localhost/items/?skip=0&limit=10。 FastAPI 通过函数的参数名自动识别它们。你可以使用 Query 函数来为这些参数添加额外的验证规则。
from fastapi import FastAPI, Query # 导入FastAPI类用于创建应用,导入Query用于定义和验证查询参数
app = FastAPI() # 创建一个FastAPI应用实例
# --- 1. 带默认值的查询参数 ---
@app.get('/item1')
def item1(item_id=Query(123)):
# 路径: /item1 或 /item1?item_id=...
# Query(123) 表示如果请求中没有提供item_id,就使用默认值123
return {'item_id': item_id}
# --- 2. 必填的查询参数 ---
@app.get('/item2')
def item2(item_id=Query(...)):
# 路径: /item2?item_id=...
# Query(...) 中的 Ellipsis(...) 表示 item_id 是一个必填的查询参数
return {'item_id': item_id}
# --- 3. 验证字符串最小长度 ---
@app.get('/item3')
def item3(item_id=Query(..., min_length=3)):
# 路径: /item3?item_id=...
# min_length=3 要求 item_id 的字符串长度必须至少为3
return {'item_id': item_id}
# --- 4. 验证字符串最小和最大长度 ---
@app.get('/item4')
def item4(item_id=Query(..., min_length=3, max_length=7)):
# 路径: /item4?item_id=...
# 同时要求 item_id 的字符串长度在3到7之间(包含3和7)
return {'item_id': item_id}
# --- 5. 验证整数范围 ---
@app.get('/item5')
def item5(item_id: int = Query(..., gt=0, lt=100)):
# 路径: /item5?item_id=...
# item_id: int 确保 item_id 是一个整数
# gt=0(greater than)要求值大于0
# lt=100(less than)要求值小于100
return {'item_id': item_id}
# --- 6. 使用别名 ---
@app.get('/item6')
def item6(item_id=Query(..., alias='id')):
# 路径: /item6?id=...
# alias='id' 允许我们使用 URL 中的别名 'id' 来对应函数中的变量名 'item_id'
return {'item_id': item_id}
# --- 7. 添加描述信息 ---
@app.get('/item7')
def item7(item_id=Query(..., description="这个字段是用来筛选产品")):
# 路径: /item7?item_id=...
# description 参数会在自动生成的API文档(如Swagger UI)中显示,用于解释参数用途
return {'item_id': item_id}
# --- 8. 标记为已弃用 ---
@app.get('/item8')
def item8(item_id=Query(..., deprecated=True)):
# 路径: /item8?item_id=...
# deprecated=True 会在API文档中标记此参数为已弃用,建议开发者停止使用
return {'item_id': item_id}
# --- 启动服务器 ---
if __name__ == '__main__':
import uvicorn # 导入uvicorn,一个ASGI服务器
uvicorn.run(
"demo07:app", # 指定要运行的应用实例,格式为 "模块名:应用实例名"
host='127.0.0.1', # 监听的IP地址
port=8000, # 监听的端口号
reload=True # 启用热重载,代码修改后服务器会自动重启
)
三、请求参数验证_Path方式
路径参数是 URL 路径的一部分,如 /items/123 中的 123。 路径参数默认就是必填的,但你可以使用 Path 函数来为它添加验证规则。
# 路径参数Path
from typing import Annotated # 导入Annotated,用于为类型添加元数据(如验证器)
from pydantic import BeforeValidator # 导入BeforeValidator,用于在Pydantic验证前执行自定义验证函数
from enum import Enum # 导入Enum,用于创建枚举类型
from fastapi import FastAPI, Path # 导入FastAPI用于创建Web应用,导入Path用于定义和验证路径参数
app = FastAPI() # 创建一个FastAPI应用实例
# --- 1. 基本路径参数 ---
@app.get("/items1/{item_id}")
def read_item1(item_id: int):
# 路径: /items1/{item_id}
# item_id 会被自动解析为整数类型。如果传入的不是整数,FastAPI会返回422错误。
return {'item_id': item_id}
# --- 2. 显式声明必填路径参数 ---
@app.get('/items2/{item_id}')
def read_item2(item_id: int = Path(...)):
# 路径: /items2/{item_id}
# Path(...) 显式声明 item_id 是一个必填的路径参数。
# 即使路径参数默认就是必填的,这样写可以增加代码的可读性,并允许添加更多验证。
return {'item_id': item_id}
# --- 3. 路径参数的数值范围验证 ---
@app.get('/items3/{item_id}')
def read_item3(item_id: int = Path(..., lt=100, gt=18)):
# 路径: /items3/{item_id}
# lt=100 (less than) 要求 item_id 小于100
# gt=18 (greater than) 要求 item_id 大于18
# 如果不满足条件,FastAPI会返回422错误。
return {'item_id': item_id}
# --- 4. 路径参数的正则表达式验证 ---
@app.get('/items4/{item_id}')
def read_item4(item_id: str = Path(..., pattern=r'^a\d{2}$')):
# 路径: /items4/{item_id}
# pattern=r'^a\d{2}$' 要求 item_id 必须符合正则表达式:
# ^a:必须以字母 'a' 开头
# \d{2}:后面必须跟着两位数字
# $:必须以两位数字结尾
# 例如:a01, a99 都是有效的,b01 或 a0 都是无效的。
return {'item_id': item_id}
# --- 5. 使用枚举类型作为路径参数 ---
class ModelName(str, Enum):
# 定义一个枚举类,继承自 str 和 Enum
# 枚举成员的值就是路径参数的有效值
alexnet = 'alexnet'
resnet = 'resnet'
lenet = 'lenet'
@app.get('/items5/{model}')
def read_item5(model: ModelName):
# 路径: /items5/{model}
# model 参数的类型被限制为 ModelName 枚举中的一个值。
# 如果传入的值不在枚举中,FastAPI会返回422错误。
return {'model': model}
# --- 6. 自定义验证器(使用Annotated和BeforeValidator) ---
# 定义一个自定义验证函数
def validate(value):
# 检查值是否以 'P-' 开头
if not value.startswith('P-'):
# 如果不符合条件,则抛出ValueError
raise ValueError('必须以P-开头')
# 如果通过验证,则返回原值
return value
# 使用Annotated和BeforeValidator创建带自定义验证的类型别名
# Item 现在是一个字符串类型,但在实际使用前会先通过 validate 函数进行验证
Item = Annotated[str, BeforeValidator(validate)]
@app.get('/items6/{item_id}')
def read_item6(item_id: Item):
# 路径: /items6/{item_id}
# item_id 会在被函数接收之前,先经过我们自定义的 validate 函数验证。
# 如果验证失败(即 validate 函数抛出ValueError),FastAPI会返回422错误。
return {'item_id': item_id}
# --- 启动服务器 ---
if __name__ == '__main__':
import uvicorn # 导入uvicorn,一个ASGI服务器
uvicorn.run(
app='demo08:app', # 指定要运行的FastAPI应用实例
host='127.0.0.1', # 服务器监听的IP地址
port=8000, # 服务器监听的端口
reload=True # 启用热重载,当代码文件发生变化时,服务器会自动重启
)
四、请求参数验证_Field方式
Field 并不是一个独立的请求方式,它是一个来自 Pydantic 的函数。它的作用是为 Pydantic 模型内部的字段添加更多的验证和元数据。
# Field验证方式
from enum import Enum # 导入Enum,用于创建枚举类型
from pydantic import field_validator # 导入field_validator,用于自定义字段验证
from fastapi import FastAPI # 导入FastAPI,用于创建Web应用
from pydantic import BaseModel, Field # 导入BaseModel用于定义数据模型,Field用于为模型字段添加验证和元数据
app = FastAPI() # 创建一个FastAPI应用实例
# --- 1. Field 必填和默认值 ---
class User(BaseModel):
# 使用Field()来为模型字段提供更多选项
name: str = Field(default='吕布') # 字段有默认值,因此是可选的
age: int = Field(...) # 使用 Ellipsis(...),表示该字段是必填的
@app.post('/users/')
def create_user(user: User):
# FastAPI会自动根据User模型验证请求体
return user
# --- 2. Field 数值范围验证 ---
class Product(BaseModel):
# gt=0(大于0),le=1000(小于等于1000)
# description='价格' 用于在自动文档中提供描述
price: float = Field(..., gt=0, le=1000, description='价格')
@app.post('/products/')
def create_product(product: Product):
return product
# --- 3. Field 字符串长度和正则验证 ---
class Account(BaseModel):
# min_length=3,max_length=20:用户名长度必须在3到20之间
username: str = Field(..., min_length=3, max_length=20)
# pattern:使用正则表达式验证密码格式
password: str = Field(..., pattern=r'^\w{6,}$') # \w表示任意字母、数字或下划线,{6,}表示至少6个
@app.post('/accounts/')
def create_account(account: Account):
return account
# --- 4. Field 添加元数据(用于文档) ---
class Item(BaseModel):
# title:在文档中显示的标题
# description:在文档中显示的详细描述
# example:在文档中显示的示例值
name: str = Field(
...,
title='商品名称',
description="必填,长度不要超过50字符",
example='手机'
)
@app.post('/items/')
def create_item(item: Item):
return item
# --- 5. 自定义验证器 ---
class User2(BaseModel):
email: str
# 使用@field_validator装饰器来定义自定义的字段验证函数
# 'email' 指定要验证的字段名
@field_validator('email')
def email_validator(cls, v):
# 验证邮箱是否包含@符号
if '@' not in v:
# 如果不满足条件,抛出ValueError,FastAPI会自动捕获并返回422错误
raise ValueError('邮箱格式错误')
return v # 验证通过,返回字段值
@app.post('/users2/')
def create_user2(user: User2):
return user
# --- 6. 验证列表长度 ---
class Order(BaseModel):
# min_items=1:列表最少包含1个元素
items: list = Field(..., min_items=1)
address: str = Field(..., description="配送地址")
@app.post('/orders/')
def create_order(order: Order):
return order
# --- 7. 使用枚举作为字段类型和默认值 ---
class Status(str, Enum):
# 定义一个枚举类
ACTIVE = 'active'
INACTIVE = 'inactive'
class Task(BaseModel):
# 使用枚举作为类型,并设置默认值
status: Status = Field(default=Status.ACTIVE)
@app.post('/tasks/')
def get_task():
# 返回一个默认的Task实例
return Task()
# --- 启动服务器 ---
if __name__ == '__main__':
import uvicorn # 导入uvicorn,一个ASGI服务器
uvicorn.run(
app='demo09:app', # 指定要运行的应用实例
host='127.0.0.1', # 服务器监听的IP地址
port=8000, # 服务器监听的端口
reload=True # 启用热重载,当代码文件发生变化时,服务器会自动重启
)

更多推荐


所有评论(0)