FastAPI 表单参数详解
表单参数是一种通过 HTML 表单提交的数据,在 FastAPI 中使用Form类来处理。表单参数通常用于处理用户通过表单提交的数据,如登录表单、注册表单等。在 FastAPI 中,使用Form2.2 Formraise ValueError('两次输入的密码不一致')return vreturn {"message": "注册成功"}# 预处理表单数据# 处理表单数据# 将处理后的数据注入到请求
1. 表单参数基础概念
1.1 什么是表单参数
表单参数是一种通过 HTML 表单提交的数据,在 FastAPI 中使用 Form 类来处理。表单参数通常用于处理用户通过表单提交的数据,如登录表单、注册表单等。
1.2 参数的区别
| 参数类型 | 传输方式 | 适用场景 | 处理方式 |
|---|---|---|---|
| 路径参数 | URL 路径 | 资源标识 | Path |
| 查询参数 | URL 查询字符串 | 过滤、分页 | Query |
| 请求体 | 请求体(JSON) | 复杂数据结构 | Body |
| 表单参数 | 请求体(表单格式) | 表单提交 | Form |
| Cookie 参数 | Cookie | 会话管理 | Cookie |
| Header 参数 | HTTP 头部 | 认证、元数据 | Header |
1.3 表单数据的传输格式
表单数据有两种主要传输格式:
- application/x-www-form-urlencoded:适用于普通表单数据,数据以键值对形式编码
- multipart/form-data:适用于包含文件上传的表单数据
1.4 表单数据的编码与解码
当表单提交时,数据会被编码为键值对形式:
username=admin&password=secret&remember=true
FastAPI 会自动解析这些数据,并根据类型注解进行类型转换。
1.5 适用场景分析
表单参数适用于以下场景:
- 用户登录和注册
- 表单提交
- 文件上传
- 简单数据提交
2. 基本表单参数实现
2.1 基本表单参数的定义方法
在 FastAPI 中,使用 Form 类来定义表单参数:
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/login")
def login(username: str = Form(...), password: str = Form(...)):
return {"username": username, "password": "******"}
2.2 Form 类的参数详解
| 参数 | 类型 | 描述 | 示例 |
|---|---|---|---|
| default | Any | 默认值 | default="guest" |
| alias | str | 参数别名 | alias="user_name" |
| description | str | 参数描述,支持 Markdown 格式 | description="用户名称,支持字母、数字和下划线" |
| title | str | 参数标题,用于 API 文档展示 | title="用户名" |
| examples | dict | 参数示例,支持多个示例 | examples={"value": "admin", "another": "user123"} |
| example | Any | 单个参数示例 | example="admin" |
| gt | int/float | 大于 | gt=0 |
| ge | int/float | 大于等于 | ge=18 |
| lt | int/float | 小于 | lt=100 |
| le | int/float | 小于等于 | le=99 |
| min_length | int | 最小长度 | min_length=3 |
| max_length | int | 最大长度 | max_length=50 |
| regex | str | 正则表达式 | regex="^[a-zA-Z0-9_]+$" |
| deprecated | bool | 是否废弃 | deprecated=True |
| include_in_schema | bool | 是否包含在 API schema 中 | include_in_schema=True |
| json_schema_extra | dict | 额外的 JSON schema 配置 | json_schema_extra={"example": "admin"} |
| alias_priority | int | 别名优先级 | alias_priority=1 |
| discriminator | str | 用于多态模型的判别器字段 | discriminator="type" |
| exclude | bool | 是否在序列化时排除此字段 | exclude=True |
| include | bool | 是否在序列化时包含此字段 | include=True |
| json_schema_mode | str | JSON schema 生成模式 | json_schema_mode="validation" |
| mode | str | 字段模式 | mode="validation" |
2.3 表单参数的类型注解
表单参数支持各种 Python 类型:
from fastapi import FastAPI, Form
from typing import Optional
app = FastAPI()
@app.post("/register")
def register(
username: str = Form(...),
password: str = Form(...),
age: int = Form(...),
email: str = Form(...),
is_active: bool = Form(False),
tags: list[str] = Form(None)
):
return {
"username": username,
"age": age,
"email": email,
"is_active": is_active,
"tags": tags
}
2.4 必需参数与可选参数
- 必需参数:使用
Form(...)或Form(None) - 可选参数:使用
Form(default=None)或Optional类型
2.5 默认值设置
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/submit")
def submit(
name: str = Form(...), # 必需参数
age: int = Form(18), # 有默认值的参数
email: str = Form(None) # 可选参数
):
return {"name": name, "age": age, "email": email}
3. 复杂表单参数处理
3.1 嵌套表单结构
使用 Pydantic模型处理嵌套表单结构:
from fastapi import FastAPI, Form
from pydantic import BaseModel
from typing import Optional
app = FastAPI()
class User(BaseModel):
username: str
password: str
email: Optional[str] = None
@app.post("/register")
def register(user: User = Form(...)):
return user
3.2 列表类型表单参数
from fastapi import FastAPI, Form
from typing import List
app = FastAPI()
@app.post("/submit")
def submit(
name: str = Form(...),
interests: List[str] = Form(...)
):
return {"name": name, "interests": interests}
3.3 字典类型表单参数
from fastapi import FastAPI, Form
from typing import Dict
app = FastAPI()
@app.post("/submit")
def submit(
name: str = Form(...),
settings: Dict[str, str] = Form(...)
):
return {"name": name, "settings": settings}
3.4 混合类型表单参数
from fastapi import FastAPI, Form
from typing import List, Optional, Dict
app = FastAPI()
@app.post("/submit")
def submit(
name: str = Form(...),
age: int = Form(...),
email: Optional[str] = Form(None),
interests: List[str] = Form(...),
settings: Dict[str, str] = Form(...)
):
return {
"name": name,
"age": age,
"email": email,
"interests": interests,
"settings": settings
}
3.5 复杂表单的验证策略
使用 Pydantic 模型进行复杂验证:
from fastapi import FastAPI, Form
from pydantic import BaseModel, EmailStr, Field
from typing import List, Optional
app = FastAPI()
class User(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
password: str = Field(..., min_length=8)
email: EmailStr
age: int = Field(..., ge=18)
interests: List[str] = []
@app.post("/register")
def register(user: User = Form(...)):
return user
4. 表单验证与数据清洗
4.1 使用 Pydantic 模型进行表单验证
from fastapi import FastAPI, Form, HTTPException
from pydantic import BaseModel, EmailStr, Field, field_validator
from typing import Optional
app = FastAPI()
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50, description="用户名")
password: str = Field(..., min_length=8, description="密码")
email: EmailStr = Field(..., description="邮箱")
age: int = Field(..., ge=18, description="年龄")
@field_validator('password')
def validate_password(cls, v):
if not any(char.isupper() for char in v):
raise ValueError('密码必须包含至少一个大写字母')
if not any(char.islower() for char in v):
raise ValueError('密码必须包含至少一个小写字母')
if not any(char.isdigit() for char in v):
raise ValueError('密码必须包含至少一个数字')
return v
@app.post("/register")
def register(user: UserCreate = Form(...)):
return {"message": "注册成功", "user": user}
4.2 自定义验证规则
from fastapi import FastAPI, Form, HTTPException
from pydantic import BaseModel, validator
app = FastAPI()
class User(BaseModel):
username: str
password: str
confirm_password: str
@validator('confirm_password')
def passwords_match(cls, v, values):
if 'password' in values and v != values['password']:
raise ValueError('两次输入的密码不一致')
return v
@app.post("/register")
def register(user: User = Form(...)):
return {"message": "注册成功"}
4.3 错误处理与响应
from fastapi import FastAPI, Form, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ValidationError
app = FastAPI()
class User(BaseModel):
username: str
password: str
@app.post("/login")
def login(username: str = Form(...), password: str = Form(...)):
try:
# 验证逻辑
if username != "admin" or password != "secret":
raise HTTPException(status_code=401, detail="用户名或密码错误")
return {"message": "登录成功"}
except Exception as e:
return JSONResponse(status_code=400, content={"detail": str(e)})
4.4 数据清洗与转换
from fastapi import FastAPI, Form
from pydantic import BaseModel, field_validator
app = FastAPI()
class User(BaseModel):
username: str
email: str
@field_validator('username')
def clean_username(cls, v):
return v.strip().lower()
@field_validator('email')
def clean_email(cls, v):
return v.strip().lower()
@app.post("/register")
def register(user: User = Form(...)):
return {"message": "注册成功", "user": user}
5. 文件上传与表单结合
5.1 单文件上传
from fastapi import FastAPI, Form, UploadFile, File
from typing import Optional
app = FastAPI()
@app.post("/upload")
def upload(
file: UploadFile = File(...),
description: Optional[str] = Form(None)
):
return {
"filename": file.filename,
"content_type": file.content_type,
"description": description
}
5.2 多文件上传
from fastapi import FastAPI, Form, UploadFile, File
from typing import List, Optional
app = FastAPI()
@app.post("/upload-multiple")
def upload_multiple(
files: List[UploadFile] = File(...),
description: Optional[str] = Form(None)
):
return {
"files": [
{"filename": file.filename, "content_type": file.content_type}
for file in files
],
"description": description
}
5.3 文件大小限制
from fastapi import FastAPI, UploadFile, File, HTTPException, Request
from fastapi.requests import Request
import asyncio
app = FastAPI()
@app.post("/upload")
async def upload(
request: Request,
file: UploadFile = File(...)
):
# 限制文件大小为 10MB
MAX_FILE_SIZE = 10 * 1024 * 1024 # 10MB
# 流式读取文件,避免内存溢出
contents = b""
size = 0
async for chunk in file.file:
size += len(chunk)
if size > MAX_FILE_SIZE:
raise HTTPException(status_code=413, detail="文件大小超过限制")
contents += chunk
# 处理文件
return {"filename": file.filename, "size": size}
# 另一种方式:使用中间件限制请求体大小
from fastapi.middleware.gzip import GZipMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
class RequestSizeLimitMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
if request.method in ["POST", "PUT", "PATCH"]:
# 限制请求体大小为 10MB
MAX_REQUEST_SIZE = 10 * 1024 * 1024
content_length = request.headers.get("Content-Length")
if content_length:
try:
if int(content_length) > MAX_REQUEST_SIZE:
return JSONResponse(
status_code=413,
content={"detail": "请求体大小超过限制"}
)
except ValueError:
# 处理 Content-Length 不是数字的情况
return JSONResponse(
status_code=400,
content={"detail": "Invalid Content-Length header"}
)
# 如果 Content-Length 不存在,继续处理请求
response = await call_next(request)
return response
# 应用中间件
app.add_middleware(RequestSizeLimitMiddleware)
5.4 文件类型验证
from fastapi import FastAPI, UploadFile, File, HTTPException
app = FastAPI()
@app.post("/upload")
async def upload(
file: UploadFile = File(...)
):
# 允许的文件类型
ALLOWED_EXTENSIONS = {"jpg", "jpeg", "png", "gif"}
# 检查文件扩展名
file_extension = file.filename.split(".")[-1].lower()
if file_extension not in ALLOWED_EXTENSIONS:
raise HTTPException(status_code=400, detail="不支持的文件类型")
return {"filename": file.filename, "extension": file_extension}
5.5 文件存储策略
import os
from fastapi import FastAPI, UploadFile, File
from pathlib import Path
app = FastAPI()
# 确保上传目录存在
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
@app.post("/upload")
async def upload(
file: UploadFile = File(...)
):
# 保存文件
file_path = UPLOAD_DIR / file.filename
with open(file_path, "wb") as buffer:
content = await file.read()
buffer.write(content)
return {"filename": file.filename, "path": str(file_path)}
5.6 upload_file 参数详解
UploadFile 类的属性和方法:
filename: 文件名content_type: 文件内容类型file: 文件对象read(): 读取文件内容write(data): 写入文件内容seek(offset): 移动文件指针close(): 关闭文件
5.7 表单数据与文件混合提交
from fastapi import FastAPI, Form, UploadFile, File
from typing import Optional
app = FastAPI()
@app.post("/submit")
def submit(
name: str = Form(...),
email: str = Form(...),
resume: UploadFile = File(...),
cover_letter: Optional[UploadFile] = File(None)
):
return {
"name": name,
"email": email,
"resume": resume.filename,
"cover_letter": cover_letter.filename if cover_letter else None
}
6. 混合使用
6.1 表单参数与路径参数
from fastapi import FastAPI, Form, Path
app = FastAPI()
@app.post("/users/{user_id}/update")
def update_user(
user_id: int = Path(..., description="用户ID"),
name: str = Form(...),
email: str = Form(...)
):
return {"user_id": user_id, "name": name, "email": email}
6.2 表单参数与查询参数
from fastapi import FastAPI, Form, Query
app = FastAPI()
@app.post("/submit")
def submit(
name: str = Form(...),
email: str = Form(...),
return_url: str = Query(..., description="提交后返回的URL")
):
return {"name": name, "email": email, "return_url": return_url}
6.3 表单参数与请求体
注意:FastAPI 不支持同时使用表单参数和请求体。以下是两种替代方案:
6.3.1 使用单一请求体
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Address(BaseModel):
street: str
city: str
zip_code: str
class UserRegister(BaseModel):
name: str
email: str
address: Address
@app.post("/register")
def register(user: UserRegister):
return {"name": user.name, "email": user.email, "address": user.address}
6.3.2 使用嵌套表单结构
from fastapi import FastAPI, Form
from pydantic import BaseModel
app = FastAPI()
class Address(BaseModel):
street: str
city: str
zip_code: str
class UserRegister(BaseModel):
name: str
email: str
address: Address
@app.post("/register")
def register(user: UserRegister = Form(...)):
return {"name": user.name, "email": user.email, "address": user.address}
6.4 Depends 与表单参数
from fastapi import FastAPI, Form, Depends
from typing import Optional
app = FastAPI()
def get_user_info(
username: str = Form(...),
password: str = Form(...),
remember: bool = Form(False)
):
return {"username": username, "password": password, "remember": remember}
@app.post("/login")
def login(user_info: dict = Depends(get_user_info)):
# 验证逻辑
if user_info["username"] == "admin" and user_info["password"] == "secret":
return {"message": "登录成功", "remember": user_info["remember"]}
return {"message": "登录失败"}
6.5 Body 与表单参数
| 特性 | 表单参数 | 请求体 |
|---|---|---|
| 传输格式 | application/x-www-form-urlencoded 或 multipart/form-data | application/json |
| 处理方式 | Form() | Body() 或 Pydantic 模型 |
| 适用场景 | 表单提交、文件上传 | 复杂数据结构、API 调用 |
| 数据类型 | 简单类型、文件 | 复杂嵌套结构 |
6.6 优先级与冲突处理
当同一参数名在多个位置出现时,优先级顺序:
- 路径参数
- 查询参数
- 请求体
- 表单参数
- Cookie
- Header
7. 表单处理的性能优化
7.1 表单数据解析优化
from fastapi import FastAPI, Form
from fastapi.requests import Request
app = FastAPI()
@app.post("/submit")
async def submit(request: Request):
# 直接解析表单数据
form_data = await request.form()
return {"form_data": dict(form_data)}
7.2 大型表单的处理策略
from fastapi import FastAPI, Form
from typing import Dict
app = FastAPI()
@app.post("/submit-large")
def submit_large(form_data: Dict[str, str] = Form(...)):
# 处理大型表单数据
return {"form_size": len(form_data), "keys": list(form_data.keys())}
7.3 缓存机制
from fastapi import FastAPI, Form
from functools import lru_cache
app = FastAPI()
@lru_cache(maxsize=128)
def process_form_data(username: str, email: str):
# 处理表单数据,结果会被缓存
return {"username": username, "email": email, "processed": True}
@app.post("/submit")
def submit(username: str = Form(...), email: str = Form(...)):
result = process_form_data(username, email)
return result
7.4 异步处理表单数据
from fastapi import FastAPI, Form
import asyncio
app = FastAPI()
async def process_data(data: dict):
# 模拟异步处理
await asyncio.sleep(1)
return {**data, "processed": True}
@app.post("/submit")
async def submit(username: str = Form(...), email: str = Form(...)):
form_data = {"username": username, "email": email}
result = await process_data(form_data)
return result
8. 表单参数的安全性与最佳实践
8.1 防 CSRF 攻击
from fastapi import FastAPI, Form, HTTPException, Depends
from fastapi.security import CSRFProtect
app = FastAPI()
csrf_protect = CSRFProtect()
@app.post("/submit")
def submit(
username: str = Form(...),
email: str = Form(...),
csrf_token: str = Form(...),
csrf: CSRFProtect = Depends(csrf_protect)
):
# 验证 CSRF token
csrf.verify_csrf_token(csrf_token)
return {"username": username, "email": email}
8.2 表单数据的验证与过滤
from fastapi import FastAPI, Form, HTTPException
from pydantic import BaseModel, Field, field_validator
import re
app = FastAPI()
class User(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
email: str = Field(..., regex=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+")
password: str = Field(..., min_length=8)
@field_validator('username')
def validate_username(cls, v):
if not re.match(r"^[a-zA-Z0-9_]+$", v):
raise ValueError('用户名只能包含字母、数字和下划线')
return v
@app.post("/register")
def register(user: User = Form(...)):
return {"message": "注册成功", "user": user}
8.3 敏感数据的处理
from fastapi import FastAPI, Form
from passlib.context import CryptContext
app = FastAPI()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
@app.post("/register")
def register(
username: str = Form(...),
password: str = Form(...)
):
# 哈希密码
hashed_password = pwd_context.hash(password)
# 存储哈希后的密码,而不是原始密码
return {"username": username, "password": "******", "hashed_password": hashed_password}
8.4 跨域表单提交
from fastapi import FastAPI, Form
from fastapi.middleware.cors import CORSMiddleware
import os
app = FastAPI()
# 配置 CORS
# 在开发环境中可以使用通配符
# 在生产环境中应该设置具体的域名
if os.getenv("ENVIRONMENT") == "production":
# 从环境变量获取允许的域名
allow_origins = os.getenv("ALLOWED_ORIGINS", "https://example.com,https://www.example.com").split(",")
else:
allow_origins = ["*"]
app.add_middleware(
CORSMiddleware,
allow_origins=allow_origins,
allow_credentials=True,
allow_methods=["POST", "GET", "PUT", "DELETE", "OPTIONS"],
allow_headers=["Content-Type", "Authorization"],
)
@app.post("/submit")
def submit(
username: str = Form(...),
email: str = Form(...)
):
return {"username": username, "email": email}
8.5 生产环境配置
from fastapi import FastAPI, Form
import os
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
app = FastAPI()
@app.post("/submit")
def submit(
username: str = Form(...),
email: str = Form(...)
):
# 使用环境变量配置
API_KEY = os.getenv("API_KEY")
# 处理表单数据
return {"username": username, "email": email, "api_key": API_KEY[:4] + "****"}
9. 常见问题与解决方案
9.1 表单数据丢失
问题:表单提交后数据丢失
原因:
- 表单提交方式错误(GET 而非 POST)
- 表单字段名与后端参数名不匹配
- 表单数据格式错误
解决方案:
- 确保使用 POST 方法提交表单
- 检查表单字段名与后端参数名是否一致
- 检查表单的
enctype属性是否正确设置
9.2 表单参数错误处理中间件
实现一个统一的表单参数错误处理中间件,用于捕获和处理表单参数验证错误:
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.middleware.base import BaseHTTPMiddleware
from pydantic import ValidationError
app = FastAPI()
class FormErrorHandlerMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
try:
response = await call_next(request)
return response
except ValidationError as e:
# 处理 Pydantic 验证错误
return JSONResponse(
status_code=422,
content={
"detail": [
{
"loc": error["loc"],
"msg": error["msg"],
"type": error["type"]
}
for error in e.errors()
]
}
)
except HTTPException as e:
# 处理 HTTP 异常
return JSONResponse(
status_code=e.status_code,
content={"detail": e.detail}
)
except Exception as e:
# 处理其他异常
return JSONResponse(
status_code=500,
content={"detail": "Internal server error"}
)
# 应用中间件
app.add_middleware(FormErrorHandlerMiddleware)
# 示例路由
from fastapi import Form
from pydantic import BaseModel, Field, field_validator
class User(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
email: str
password: str = Field(..., min_length=8)
@field_validator('password')
def validate_password(cls, v):
if not any(char.isupper() for char in v):
raise ValueError('密码必须包含至少一个大写字母')
if not any(char.islower() for char in v):
raise ValueError('密码必须包含至少一个小写字母')
if not any(char.isdigit() for char in v):
raise ValueError('密码必须包含至少一个数字')
return v
@app.post("/register")
def register(user: User = Form(...)):
return {"message": "注册成功", "user": user}
@app.post("/login")
def login(
username: str = Form(..., min_length=3),
password: str = Form(..., min_length=8)
):
if username == "admin" and password == "Password123":
return {"message": "登录成功"}
raise HTTPException(status_code=401, detail="用户名或密码错误")
错误处理中间件的优势:
- 统一处理表单参数验证错误
- 提供一致的错误响应格式
- 减少重复的错误处理代码
- 提高代码的可维护性
使用建议:
- 在生产环境中使用错误处理中间件
- 根据具体需求自定义错误响应格式
- 记录错误日志以便排查问题
9.3 编码问题
问题:表单提交的中文数据出现乱码
原因:
- 表单编码设置错误
- 后端解码方式不正确
解决方案:
- 在表单中设置
accept-charset="UTF-8" - 确保后端使用 UTF-8 编码处理数据
9.4 文件上传失败
问题:文件上传失败或文件大小为 0
原因:
- 文件大小超过限制
- 文件类型不支持
- 上传路径权限不足
解决方案:
- 检查文件大小限制设置
- 检查文件类型验证逻辑
- 确保上传目录存在且有写入权限
9.5 验证错误处理
问题:验证错误信息不明确
原因:
- 自定义验证器没有提供清晰的错误信息
- 错误处理逻辑不完善
解决方案:
- 在验证器中提供详细的错误信息
- 使用统一的错误处理中间件
9.6 排查与解决方法
- 检查请求头:确保
Content-Type正确设置 - 检查表单字段:确保字段名与后端参数名一致
- 检查验证逻辑:确保验证规则合理
- 检查服务器日志:查看详细的错误信息
- 使用工具调试:使用 Postman 或 curl 测试表单提交
10. 高级表单处理技巧
10.1 动态表单生成
from fastapi import FastAPI, Form
from typing import Dict, Any
app = FastAPI()
@app.post("/dynamic-form")
def dynamic_form(form_data: Dict[str, Any] = Form(...)):
# 处理动态生成的表单
return {"form_data": form_data, "fields": list(form_data.keys())}
10.2 条件验证
from fastapi import FastAPI, Form
from pydantic import BaseModel, Field, field_validator
from typing import Optional
app = FastAPI()
class User(BaseModel):
username: str
email: str
age: int
is_student: bool = False
student_id: Optional[str] = None
@field_validator('student_id')
def validate_student_id(cls, v, values):
if values.get('is_student') and not v:
raise ValueError('学生必须提供学生证号')
return v
@app.post("/register")
def register(user: User = Form(...)):
return {"message": "注册成功", "user": user}
10.3 表单数据预处理与后处理
from fastapi import FastAPI, Form
from pydantic import BaseModel, field_validator
app = FastAPI()
class User(BaseModel):
username: str
email: str
@field_validator('username', 'email', mode='before')
def preprocess(cls, v):
# 预处理:去除首尾空格
if isinstance(v, str):
return v.strip().lower()
return v
@field_validator('username', mode='after')
def postprocess(cls, v):
# 后处理:确保用户名格式正确
return v
@app.post("/register")
def register(user: User = Form(...)):
return {"message": "注册成功", "user": user}
10.4 自定义表单处理器
from fastapi import FastAPI, Form, Request
from fastapi.routing import APIRoute
from starlette.responses import Response
app = FastAPI()
class FormRoute(APIRoute):
def get_route_handler(self):
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
# 预处理表单数据
if request.method == "POST" and "application/x-www-form-urlencoded" in request.headers.get("Content-Type", ""):
form_data = await request.form()
# 处理表单数据
processed_data = {k: v.strip() for k, v in form_data.items()}
# 将处理后的数据注入到请求中
request.scope["form"] = processed_data
return await original_route_handler(request)
return custom_route_handler
app.router.route_class = FormRoute
@app.post("/submit")
def submit(username: str = Form(...), email: str = Form(...)):
return {"username": username, "email": email}
10.5 表单参数的文档生成
FastAPI 会自动为表单参数生成 API 文档,包括:
- 参数名称
- 参数类型
- 是否必需
- 默认值
- 描述
- 示例
10.6 表单参数的序列化与反序列化
from fastapi import FastAPI, Form
from pydantic import BaseModel
app = FastAPI()
class User(BaseModel):
username: str
email: str
age: int
@app.post("/register")
def register(user: User = Form(...)):
# 序列化
user_dict = user.model_dump()
# 反序列化
new_user = User(**user_dict)
return {"message": "注册成功", "user": new_user}
10.7 表单参数的国际化处理
10.7.1 基本实现
from fastapi import FastAPI, Form, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/form", response_class=HTMLResponse)
def get_form(request: Request):
return templates.TemplateResponse("form.html", {"request": request})
@app.post("/submit")
def submit(
username: str = Form(...),
email: str = Form(...),
language: str = Form("en")
):
# 根据语言返回不同的消息
messages = {
"en": "Form submitted successfully",
"zh": "表单提交成功",
"es": "Formulario enviado con éxito"
}
return {"message": messages.get(language, "Form submitted successfully"), "username": username, "email": email}
10.7.2 与 i18n 框架集成
使用 python-i18n 库实现更完整的国际化支持:
from fastapi import FastAPI, Form, Request
import i18n
# 配置 i18n
i18n.load_path = ["locales"]
i18n.set('locale', 'en')
i18n.set('fallback', 'en')
app = FastAPI()
@app.post("/submit")
def submit(
username: str = Form(...),
email: str = Form(...),
language: str = Form("en")
):
# 设置语言
i18n.set('locale', language)
# 使用翻译
message = i18n.t("form.submitted_successfully")
return {"message": message, "username": username, "email": email}
# locales/en.yml
# form:
# submitted_successfully: "Form submitted successfully"
# locales/zh.yml
# form:
# submitted_successfully: "表单提交成功"
# locales/es.yml
# form:
# submitted_successfully: "Formulario enviado con éxito"
10.7.3 自动检测语言
from fastapi import FastAPI, Form, Request
from fastapi.responses import JSONResponse
import i18n
# 配置 i18n
i18n.load_path = ["locales"]
i18n.set('fallback', 'en')
app = FastAPI()
@app.post("/submit")
def submit(
request: Request,
username: str = Form(...),
email: str = Form(...)
):
# 从请求头检测语言
accept_language = request.headers.get("Accept-Language", "en")
# 提取语言代码
language = accept_language.split(",")[0].split(";")[0]
# 设置语言
i18n.set('locale', language)
# 使用翻译
message = i18n.t("form.submitted_successfully")
return {"message": message, "username": username, "email": email, "language": language}
10.8 高级应用案例分析
10.8.1 案例:用户注册与文件上传
from fastapi import FastAPI, Form, UploadFile, File
from pydantic import BaseModel, EmailStr, Field, field_validator
from typing import Optional
import os
from pathlib import Path
app = FastAPI()
# 确保上传目录存在
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
class UserRegister(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
email: EmailStr
password: str = Field(..., min_length=8)
confirm_password: str
avatar: Optional[UploadFile] = None
@field_validator('confirm_password')
def passwords_match(cls, v, values):
if 'password' in values and v != values['password']:
raise ValueError('两次输入的密码不一致')
return v
@app.post("/register")
async def register(
username: str = Form(...),
email: str = Form(...),
password: str = Form(...),
confirm_password: str = Form(...),
avatar: Optional[UploadFile] = File(None)
):
# 验证密码
if password != confirm_password:
return {"message": "两次输入的密码不一致"}
# 处理头像上传
avatar_path = None
if avatar:
avatar_path = UPLOAD_DIR / avatar.filename
with open(avatar_path, "wb") as buffer:
content = await avatar.read()
buffer.write(content)
return {
"message": "注册成功",
"user": {
"username": username,
"email": email,
"avatar": str(avatar_path) if avatar_path else None
}
}
10.8.2 案例:多步骤表单提交
from fastapi import FastAPI, Form, Request
from fastapi.responses import RedirectResponse
from typing import Optional
from starlette.middleware.sessions import SessionMiddleware
import secrets
app = FastAPI()
# 添加会话中间件
app.add_middleware(SessionMiddleware, secret_key=secrets.token_urlsafe(32))
@app.post("/step1")
def step1(
request: Request,
name: str = Form(...),
email: str = Form(...)
):
# 存储步骤1的数据
request.session["name"] = name
request.session["email"] = email
return RedirectResponse("/step2", status_code=303)
@app.get("/step2")
def get_step2():
return {"message": "请填写步骤2"}
@app.post("/step2")
def step2(
request: Request,
age: int = Form(...),
address: str = Form(...)
):
# 存储步骤2的数据
request.session["age"] = age
request.session["address"] = address
return RedirectResponse("/step3", status_code=303)
@app.get("/step3")
def get_step3():
return {"message": "请确认信息"}
@app.post("/step3")
def step3(request: Request):
# 获取所有步骤的数据
user_data = {
"name": request.session.get("name"),
"email": request.session.get("email"),
"age": request.session.get("age"),
"address": request.session.get("address")
}
# 处理提交的数据
return {"message": "表单提交成功", "data": user_data}更多推荐



所有评论(0)