1. 表单参数基础概念

1.1 什么是表单参数

表单参数是一种通过 HTML 表单提交的数据,在 FastAPI 中使用 Form 类来处理。表单参数通常用于处理用户通过表单提交的数据,如登录表单、注册表单等。

1.2 参数的区别

参数类型传输方式适用场景处理方式
路径参数URL 路径资源标识Path
查询参数URL 查询字符串过滤、分页Query
请求体请求体(JSON)复杂数据结构Body
表单参数请求体(表单格式)表单提交Form
Cookie 参数Cookie会话管理Cookie
Header 参数HTTP 头部认证、元数据Header

1.3 表单数据的传输格式

表单数据有两种主要传输格式:

  1. application/x-www-form-urlencoded:适用于普通表单数据,数据以键值对形式编码
  2. 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 类的参数详解

参数类型描述示例
defaultAny默认值default="guest"
aliasstr参数别名alias="user_name"
descriptionstr参数描述,支持 Markdown 格式description="用户名称,支持字母、数字和下划线"
titlestr参数标题,用于 API 文档展示title="用户名"
examplesdict参数示例,支持多个示例examples={"value": "admin", "another": "user123"}
exampleAny单个参数示例example="admin"
gtint/float大于gt=0
geint/float大于等于ge=18
ltint/float小于lt=100
leint/float小于等于le=99
min_lengthint最小长度min_length=3
max_lengthint最大长度max_length=50
regexstr正则表达式regex="^[a-zA-Z0-9_]+$"
deprecatedbool是否废弃deprecated=True
include_in_schemabool是否包含在 API schema 中include_in_schema=True
json_schema_extradict额外的 JSON schema 配置json_schema_extra={"example": "admin"}
alias_priorityint别名优先级alias_priority=1
discriminatorstr用于多态模型的判别器字段discriminator="type"
excludebool是否在序列化时排除此字段exclude=True
includebool是否在序列化时包含此字段include=True
json_schema_modestrJSON schema 生成模式json_schema_mode="validation"
modestr字段模式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-dataapplication/json
处理方式Form()Body() 或 Pydantic 模型
适用场景表单提交、文件上传复杂数据结构、API 调用
数据类型简单类型、文件复杂嵌套结构

6.6 优先级与冲突处理

当同一参数名在多个位置出现时,优先级顺序:

  1. 路径参数
  2. 查询参数
  3. 请求体
  4. 表单参数
  5. Cookie
  6. 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 排查与解决方法

  1. 检查请求头:确保 Content-Type 正确设置
  2. 检查表单字段:确保字段名与后端参数名一致
  3. 检查验证逻辑:确保验证规则合理
  4. 检查服务器日志:查看详细的错误信息
  5. 使用工具调试:使用 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}
Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐