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 类的参数详解

参数 类型 描述 示例
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 优先级与冲突处理

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

  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社区

更多推荐