FastAPI 教程

FastAPI 教程

FastAPI 是一个现代、高性能的 Python Web 框架，用于构建 RESTful API。它基于 Starlette（性能）与 Pydantic（数据校验），以开发速度快、性能高、自带文档著称，是目前 Python Web 领域最热门的框架之一。

FastAPI 的核心优势

优势	说明
高性能	与 Node.js、Go 相当的并发性能
自动文档	自动生成 Swagger UI 交互式文档
类型校验	基于类型注解自动校验请求数据
代码简洁	声明式开发，样板代码少
异步支持	原生 async/await，天然支持高并发

一个最简应用

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return {"message": "Hello CodeNest"}

# 启动服务
uvicorn main:app --reload
# 访问 http://127.0.0.1:8000

为什么选 FastAPI

• 自动文档：访问 /docs 即可在线调试接口
• 类型安全：IDE 提示完善，错误提前发现
• 社区活跃：生态完善（SQLAlchemy、Pydantic 深度集成）
• 团队友好：OpenAPI 规范自动导出，前后端协作顺畅

适用场景

• 构建 RESTful API 后端
• 前后端分离项目
• 微服务
• 快速原型开发
• 机器学习模型服务化

FastAPI 安装

安装 FastAPI 非常简单。本节介绍环境准备、安装命令和版本要求。

环境要求

• Python 3.8+（推荐 3.10+）
• pip（Python 包管理器）
• 建议使用虚拟环境（venv）

创建虚拟环境（推荐）

# 创建虚拟环境
python3 -m venv venv

# 激活（macOS/Linux）
source venv/bin/activate

# 激活（Windows）
venv\Scripts\activate

安装 FastAPI

# 基础安装
pip install fastapi

# 完整安装（含 uvicorn 服务器）
pip install "fastapi[standard]"

安装 Uvicorn（ASGI 服务器）

pip install "uvicorn[standard]"

验证安装

python -c "import fastapi; print(fastapi.__version__)"
# 例如输出：0.115.0

安装 SQLAlchemy（数据库）

pip install sqlalchemy pymysql

保存依赖

pip freeze > requirements.txt

常见问题

问题	解决
pip 版本过低	python -m pip install -U pip
下载慢	使用国内镜像：pip install fastapi -i https://pypi.tuna.tsinghua.edu.cn/simple
提示 Python 版本低	升级到 Python 3.8+

VSCode FastAPI

在 VS Code 中配置 FastAPI 开发环境，获得智能提示、调试和自动补全体验。

1. 安装 Python 扩展

VS Code 扩展市场搜索并安装：

• Python（微软官方，必备）
• Pylance（类型检查与智能提示）
• Ruff（代码检查与格式化）
• Thunder Client（接口调试）

2. 选择 Python 解释器

Ctrl+Shift+P → Python: Select Interpreter
选择虚拟环境 venv 中的 Python

3. 创建项目结构

myapi/
├── venv/          # 虚拟环境
├── main.py        # 入口文件
└── requirements.txt

4. 配置调试

创建 .vscode/launch.json：

{
  "version": "0.2.0",
  "configurations": [
    {
      "name": "FastAPI",
      "type": "debugpy",
      "request": "launch",
      "module": "uvicorn",
      "args": ["main:app", "--reload"],
      "jinja": true
    }
  ]
}

5. 类型提示体验

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items")
def create_item(item: Item):
    # Pylance 会自动提示 item.name / item.price
    return {"name": item.name, "total": item.price * 2}

6. 运行与调试

• 终端运行：uvicorn main:app --reload
• 或按 F5 启动调试，设置断点逐步排查

第一个 FastAPI 应用

创建你的第一个 FastAPI 应用，从零走通"写代码 → 启动 → 访问"全流程。

第一步：编写代码

新建 main.py：

from fastapi import FastAPI

# 创建应用实例
app = FastAPI()

# 定义根路由
@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!"}

# 定义带路径参数的路由
@app.get("/hello/{name}")
def say_hello(name: str):
    return {"message": f"Hello, {name}!"}

第二步：启动服务

uvicorn main:app --reload

# 输出：
# Uvicorn running on http://127.0.0.1:8000

第三步：访问测试

# 终端请求
curl http://127.0.0.1:8000/
# {"message":"Hello, FastAPI!"}

curl http://127.0.0.1:8000/hello/CodeNest
# {"message":"Hello, CodeNest!"}

第四步：打开自动文档

浏览器访问：

• http://127.0.0.1:8000/docs → Swagger UI 交互文档
• http://127.0.0.1:8000/redoc → ReDoc 文档

代码逐行解析

代码	作用
FastAPI()	创建应用实例
@app.get("/")	注册 GET 路由
def read_root()	视图函数，返回 JSON
return {"message": ...}	自动转为 JSON 响应
{name}	路径参数，自动提取

FastAPI 交互式 API 文档

FastAPI 最酷的特性之一：自动生成交互式 API 文档，无需任何额外配置。

两种文档

文档	地址	特点
Swagger UI	/docs	交互式，可直接调试接口
ReDoc	/redoc	更整洁的参考文档
OpenAPI JSON	/openapi.json	机器可读的规范文件

Swagger UI 的功能

• 列出所有接口（方法、路径、参数）
• 点击"Try it out"直接发送请求
• 展示请求/响应模型结构
• 支持认证配置

自定义文档元数据

from fastapi import FastAPI

app = FastAPI(
    title="CodeNest API",
    description="个人知识库网站的后端接口文档",
    version="1.0.0",
    docs_url="/api/docs",     # 自定义 Swagger 地址
    redoc_url="/api/redoc",   # 自定义 ReDoc 地址
)

为接口添加说明

@app.get("/users", tags=["用户"], summary="获取用户列表")
def get_users():
    """
    返回系统中的所有用户。

    - 支持分页
    - 按创建时间倒序
    """
    return [{"id": 1, "name": "张三"}]

文档会展示 tags 分组和 summary 说明。

文档的用途

• 前端开发：对照文档联调接口
• 团队协作：接口规范一目了然
• 自动化测试：基于 OpenAPI 生成测试
• 代码生成：生成客户端 SDK

FastAPI 基本路由

本节掌握 FastAPI 的基本路由：HTTP 方法、路径定义、动态参数。

HTTP 方法路由

from fastapi import FastAPI

app = FastAPI()

@app.get("/items")          # 查询
def get_items():
    return {"method": "GET"}

@app.post("/items")         # 创建
def create_item():
    return {"method": "POST"}

@app.put("/items/{id}")     # 更新
def update_item(id: int):
    return {"method": "PUT", "id": id}

@app.delete("/items/{id}")  # 删除
def delete_item(id: int):
    return {"method": "DELETE", "id": id}

路径参数

@app.get("/users/{user_id}")
def get_user(user_id: int):      # 类型注解自动转换+校验
    return {"user_id": user_id}

# 访问 /users/42 → {"user_id": 42}
# 访问 /users/abc → 422 校验错误

多段路径参数

@app.get("/users/{user_id}/posts/{post_id}")
def get_post(user_id: int, post_id: int):
    return {"user_id": user_id, "post_id": post_id}

路由顺序

静态路径要定义在动态路径之前：

# 正确：先静态后动态
@app.get("/users/me")
def get_me():
    return {"user": "me"}

@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

请求方法别名

# 用 api 装饰器可指定多个方法
from fastapi import FastAPI

app = FastAPI()

@app.api_route("/ping", methods=["GET", "POST"])
def ping():
    return {"pong": True}

FastAPI 请求和响应

FastAPI 自动处理请求与响应的序列化，你只需关注业务逻辑。

自动 JSON 响应

@app.get("/data")
def get_data():
    # 返回字典自动转 JSON
    return {"name": "CodeNest", "count": 100}

返回任意类型

from fastapi import FastAPI

app = FastAPI()

@app.get("/list")
def get_list():
    return [1, 2, 3]           # 列表

@app.get("/text")
def get_text():
    return "plain text"        # 字符串

@app.get("/none")
def get_none():
    return None                # null

自定义状态码

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items", status_code=status.HTTP_201_CREATED)
def create_item():
    return {"id": 1}

@app.delete("/items/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(id: int):
    return  # 204 无内容

响应头设置

from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/custom")
def custom(response: Response):
    response.headers["X-Custom"] = "value"
    return {"ok": True}

JSONResponse 直接返回

from fastapi.responses import JSONResponse

@app.get("/raw")
def raw():
    return JSONResponse(
        content={"status": "ok"},
        status_code=200,
        headers={"X-Version": "1.0"},
    )

响应模型

from pydantic import BaseModel

class UserOut(BaseModel):
    id: int
    name: str

@app.get("/users/{uid}", response_model=UserOut)
def get_user(uid: int):
    # 只返回 UserOut 中定义的字段
    return {"id": uid, "name": "张三", "secret": "隐藏"}

FastAPI Pydantic 模型

Pydantic 模型是 FastAPI 数据校验的核心。用类型注解定义数据结构，FastAPI 自动完成解析、校验和文档生成。

定义模型

from pydantic import BaseModel

class Item(BaseModel):
    name: str                    # 必填字符串
    price: float                 # 必填浮点数
    is_offer: bool = False       # 可选，默认 False
    tags: list[str] = []         # 字符串列表，默认空

作为请求体使用

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items")
def create_item(item: Item):
    # 请求体会自动校验并转换为 Item 对象
    return {"name": item.name, "price": item.price}

字段校验

from pydantic import BaseModel, Field

class User(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    age: int = Field(ge=0, le=150)         # 0-150
    email: str = Field(pattern=r"^\S+@\S+\.\S+$")
    score: float = Field(default=0, ge=0)

校验失败行为

// 请求体不合法时返回 422：
{
  "detail": [
    {
      "loc": ["body", "age"],
      "msg": "Input should be greater than or equal to 0",
      "type": "greater_than_equal"
    }
  ]
}

嵌套模型

class Address(BaseModel):
    city: str
    zipcode: str

class User(BaseModel):
    name: str
    address: Address          # 嵌套模型

# 请求体：
# {"name": "张三", "address": {"city": "北京", "zipcode": "100000"}}

模型转字典/JSON

item = Item(name="书", price=29.9)
print(item.model_dump())      # {'name': '书', 'price': 29.9}
print(item.model_dump_json())  # '{"name":"书","price":29.9}'

FastAPI 依赖注入

依赖注入（Dependency Injection）是 FastAPI 最强大的特性之一，它让代码更模块化、可复用、易测试。

什么是依赖注入

把"函数需要的依赖"通过参数声明，由 FastAPI 自动解析和提供，而不是在函数内部手动创建。

最简单示例

from fastapi import FastAPI, Depends

app = FastAPI()

def get_db():
    # 模拟数据库连接
    return {"connected": True}

@app.get("/users")
def get_users(db=Depends(get_db)):
    # db 由 FastAPI 自动注入
    return {"db": db, "users": []}

依赖的依赖

def get_db():
    return "数据库连接"

def get_current_user(db=Depends(get_db)):
    return {"user": "张三", "db": db}

@app.get("/me")
def read_me(user=Depends(get_current_user)):
    return user

类作为依赖

class Pagination:
    def __init__(self, page: int = 1, page_size: int = 10):
        self.page = page
        self.page_size = page_size

@app.get("/items")
def get_items(pag: Pagination = Depends()):
    return {"page": pag.page, "size": pag.page_size}

实际场景：数据库会话

from sqlalchemy.orm import Session
from .database import SessionLocal

def get_db():
    db = SessionLocal()
    try:
        yield db          # yield 形式：请求结束后自动关闭
    finally:
        db.close()

@app.get("/users")
def get_users(db: Session = Depends(get_db)):
    return db.query(User).all()

依赖注入的好处

• 复用：认证、日志、数据库等逻辑只写一次
• 测试：测试时可替换依赖（mock）
• 清晰：函数依赖一目了然
• 性能：依赖可缓存、按需创建

FastAPI 表单数据

处理表单数据（application/x-www-form-urlencoded 或 multipart/form-data）时，需要使用 Form 参数。

表单与 JSON 的区别

对比项	JSON Body	Form 表单
Content-Type	application/json	application/x-www-form-urlencoded
定义方式	BaseModel	Form() 参数
适用	API 接口	传统表单、OAuth2

基本表单

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login")
def login(username: str = Form(), password: str = Form()):
    return {"username": username, "logged_in": True}

curl -X POST http://127.0.0.1:8000/login   -d "username=admin&password=123456"

带默认值的表单字段

@app.post("/settings")
def save_settings(
    theme: str = Form("dark"),
    lang: str = Form("zh-CN"),
):
    return {"theme": theme, "lang": lang}

表单校验

from fastapi import Form
from pydantic import Field

@app.post("/register")
def register(
    username: str = Form(min_length=3, max_length=20),
    password: str = Form(min_length=6),
):
    return {"ok": True}

文件上传（multipart）

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/upload")
async def upload(file: UploadFile):
    content = await file.read()      # 读取文件内容
    return {
        "filename": file.filename,
        "size": len(content),
        "content_type": file.content_type,
    }

表单 + 文件混合

@app.post("/post")
async def create_post(
    title: str = Form(),
    image: UploadFile = File(None),   # 可选文件
):
    return {"title": title, "image": image.filename if image else None}

FastAPI 核心概念

本节总结 FastAPI 的核心概念，帮你建立完整的框架知识地图。

FastAPI 的技术栈

组件	作用
Starlette	底层 ASGI 框架（路由、中间件、WebSocket）
Pydantic	数据模型与校验
Uvicorn	ASGI 服务器
OpenAPI	API 规范（自动生成文档）

请求处理流程

1. 请求到达 Uvicorn
2. Starlette 路由匹配到端点函数
3. FastAPI 按声明解析参数（路径/查询/请求体）
4. Pydantic 校验数据（失败返回 422）
5. 执行依赖注入
6. 执行端点函数，返回结果
7. 结果序列化（response_model 过滤）返回客户端

声明式开发模式

@app.get("/items/{item_id}")
def get_item(
    item_id: int,                    # 路径参数
    q: str | None = None,            # 查询参数（可选）
    item: Item | None = None,        # 请求体（可选）
):
    return {"item_id": item_id, "q": q}

参数来源由声明位置决定：路径中→路径参数；查询串→查询参数；模型类型→请求体。

异步支持

# 同步端点（FastAPI 自动放入线程池）
@app.get("/sync")
def sync_endpoint():
    return {"type": "sync"}

# 异步端点（适合 IO 密集）
@app.get("/async")
async def async_endpoint():
    await some_io_operation()
    return {"type": "async"}

FastAPI 与其他框架对比

框架	风格	性能	文档
FastAPI	类型化/异步	高	自动
Flask	简单/同步	中	手动
Django	全栈/重量	中	手动

FastAPI 路径参数

路径参数是 URL 中动态变化的部分（如 /users/42 中的 42），用于定位具体资源。

基本用法

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

类型转换与校验

声明类型后 FastAPI 自动转换并校验：

@app.get("/items/{item_id}")
def get_item(item_id: int):     # 非数字 → 422
    return {"item_id": item_id}

# /items/42 → {"item_id": 42}
# /items/abc → 422 Unprocessable Entity

字符串路径参数

@app.get("/files/{file_name:path}")
def get_file(file_name: str):
    # :path 允许包含斜杠的多段路径
    return {"file": file_name}

# /files/docs/guide.pdf → file_name="docs/guide.pdf"

多个路径参数

@app.get("/categories/{cat_id}/posts/{post_id}")
def get_post(cat_id: int, post_id: int):
    return {"cat_id": cat_id, "post_id": post_id}

路径参数校验

from fastapi import Path

@app.get("/users/{user_id}")
def get_user(
    user_id: int = Path(ge=1, description="用户ID，从1开始"),
):
    return {"user_id": user_id}

枚举路径参数

from enum import Enum

class Level(str, Enum):
    easy = "easy"
    hard = "hard"

@app.get("/game/{level}")
def play(level: Level):
    # 只接受 easy / hard
    return {"level": level.value}

路径参数 vs 查询参数

对比项	路径参数	查询参数
位置	URL 路径中	? 后面
示例	/users/42	/users?page=2
用途	定位资源	过滤/分页/排序

FastAPI 查询参数

查询参数是 URL 中 ? 后面的键值对（如 ?page=2&size=10），用于过滤、分页、排序等。

基本用法

from fastapi import FastAPI

app = FastAPI()

@app.get("/items")
def get_items(page: int = 1, size: int = 10):
    # 声明在函数中、不在路径里的参数即查询参数
    return {"page": page, "size": size}

# /items?page=2&size=20 → {"page": 2, "size": 20}
# /items → 使用默认值 {"page": 1, "size": 10}

可选查询参数

@app.get("/search")
def search(q: str | None = None, category: str | None = None):
    return {"q": q, "category": category}

# /search → {"q": None, "category": None}
# /search?q=python → {"q": "python", "category": None}

布尔与数字

@app.get("/posts")
def get_posts(
    published: bool = False,   # ?published=true
    limit: int = 10,           # ?limit=50
    rate: float = 0.5,         # ?rate=0.8
):
    return {"published": published, "limit": limit, "rate": rate}

多个值（列表）

from typing import Annotated

@app.get("/tags")
def filter_tags(tags: Annotated[list[str], Query()] = []):
    return {"tags": tags}

# /tags?tags=python&tags=ai → {"tags": ["python", "ai"]}

必填查询参数

@app.get("/weather")
def weather(city: str):   # 无默认值 → 必填
    return {"city": city}

# /weather?city=beijing → {"city": "beijing"}
# /weather → 422 错误

完整示例

@app.get("/api/tutorials")
def list_tutorials(
    page: int = 1,
    page_size: int = 10,
    keyword: str | None = None,
    category: str | None = None,
):
    # 实际查询逻辑...
    return {
        "page": page,
        "page_size": page_size,
        "keyword": keyword,
        "category": category,
    }

FastAPI 请求体

请求体（Request Body）是请求中携带的数据，通常为 JSON。在 FastAPI 中用 Pydantic 模型声明。

基本请求体

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.post("/items")
def create_item(item: Item):
    # 请求体自动校验并转换为 Item 实例
    return {"item": item.model_dump()}

curl -X POST http://127.0.0.1:8000/items   -H "Content-Type: application/json"   -d '{"name": "键盘", "price": 299}'

请求体 + 路径参数

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item.model_dump()}

请求体 + 查询参数

@app.put("/items/{item_id}")
def update_item(
    item_id: int,
    item: Item,
    q: str | None = None,       # 查询参数
):
    return {"item_id": item_id, "q": q, "item": item.model_dump()}

多个请求体参数

class User(BaseModel):
    username: str

class Item2(BaseModel):
    name: str

@app.put("/combine")
def combine(user: User, item: Item2):
    # 请求体：{"user": {...}, "item": {...}}
    return {"user": user.username, "item": item.name}

嵌套请求体

class Image(BaseModel):
    url: str
    name: str

class Item3(BaseModel):
    name: str
    image: Image | None = None
    tags: list[str] = []

请求体可选

@app.post("/search")
def search(item: Item | None = None):
    # 请求体可选，不传也能调用
    if item:
        return {"searched": item.name}
    return {"searched": "全部"}

FastAPI 查询参数校验

对查询参数进行校验：长度、范围、正则等，让接口更健壮。

基础校验

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items")
def get_items(
    q: str | None = Query(default=None, max_length=50),
    page: int = Query(default=1, ge=1),
):
    return {"q": q, "page": page}

常用校验参数

参数	含义
min_length / max_length	字符串长度范围
ge / le / gt / lt	数值范围（≥ ≤ > <）
pattern	正则校验
min_items / max_items	列表长度范围
description	文档说明
example	示例值（文档展示）
deprecated	标记废弃参数

字符串校验

@app.get("/search")
def search(
    q: str = Query(min_length=2, max_length=100, description="搜索关键词"),
):
    return {"q": q}

正则校验

@app.get("/code")
def check_code(
    code: str = Query(pattern=r"^[A-Z]{2}\d{4}$", description="格式如 AB1234"),
):
    return {"code": code}

数值范围校验

@app.get("/filter")
def filter_items(
    min_price: float = Query(ge=0),
    max_price: float = Query(le=10000),
    rating: float = Query(default=4.5, ge=0, le=5),
):
    return {"min_price": min_price, "max_price": max_price}

枚举查询参数

@app.get("/posts")
def get_posts(
    order: str = Query(default="desc", pattern="^(asc|desc)$"),
):
    return {"order": order}

带示例的查询参数

@app.get("/items")
def get_items(
    page: int = Query(default=1, ge=1, example=1),
    page_size: int = Query(default=10, ge=1, le=100, example=20),
):
    return {"page": page, "page_size": page_size}

FastAPI 路径参数数值校验

对路径参数进行数值校验，如 ID 必须为正整数。

基本用法

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
def get_item(
    item_id: int = Path(ge=1, description="商品 ID"),
):
    return {"item_id": item_id}

# /items/1 → OK
# /items/0 → 422（必须 ≥ 1）
# /items/-5 → 422

完整校验参数

@app.get("/users/{user_id}")
def get_user(
    user_id: int = Path(
        ge=1,               # ≥ 1
        le=1000000,         # ≤ 1000000
        description="用户 ID",
    ),
):
    return {"user_id": user_id}

多参数校验

@app.get("/posts/{post_id}/comments/{comment_id}")
def get_comment(
    post_id: int = Path(ge=1),
    comment_id: int = Path(ge=1),
):
    return {"post_id": post_id, "comment_id": comment_id}

字符串路径参数校验

@app.get("/users/{username}")
def get_user(
    username: str = Path(min_length=3, max_length=20),
):
    return {"username": username}

路径参数与查询参数混合

@app.get("/items/{item_id}")
def get_item(
    item_id: int = Path(ge=1),
    q: str | None = Query(default=None, max_length=50),
):
    return {"item_id": item_id, "q": q}

校验失败的响应

// GET /items/0 → 422
{
  "detail": [
    {
      "loc": ["path", "item_id"],
      "msg": "Input should be greater than or equal to 1",
      "type": "greater_than_equal"
    }
  ]
}

FastAPI 请求体字段与嵌套模型

为请求体字段添加校验，并构建嵌套模型表达复杂数据结构。

字段级校验（Field）

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    price: float = Field(gt=0, description="价格必须大于0")
    quantity: int = Field(ge=0, default=1)
    email: str | None = Field(default=None, pattern=r"^\S+@\S+$")

嵌套模型

class Address(BaseModel):
    city: str
    street: str
    zipcode: str = Field(pattern=r"^\d{6}$")

class User(BaseModel):
    name: str
    age: int = Field(ge=0, le=150)
    address: Address          # 嵌套
    tags: list[str] = []      # 数组

// 合法请求体：
{
  "name": "张三",
  "age": 25,
  "address": {"city": "北京", "street": "中关村", "zipcode": "100000"},
  "tags": ["python", "ai"]
}

列表嵌套

class OrderItem(BaseModel):
    product_id: int
    quantity: int = Field(ge=1)

class Order(BaseModel):
    user_id: int
    items: list[OrderItem]     # 对象数组
    note: str | None = None

可选字段与默认值

class Config(BaseModel):
    theme: str = "dark"                  # 默认值
    notifications: bool = True           # 默认值
    extra: dict | None = None            # 可选字典
    callback: str | None = None          # 可选

数据校验失败示例

// 提交 age = -5 → 422
{
  "detail": [{
    "loc": ["body", "age"],
    "msg": "Input should be greater than or equal to 0",
    "type": "greater_than_equal"
  }]
}

嵌套校验深度

嵌套模型会逐层校验：最内层的 zipcode 不合法也会报错，并指出完整的字段路径（loc: ["body", "address", "zipcode"]）。

FastAPI 响应模型

响应模型（response_model）定义接口返回的数据结构，用于过滤字段、保证一致性、完善文档。

基本用法

from pydantic import BaseModel

class UserOut(BaseModel):
    id: int
    name: str

class UserIn(BaseModel):
    id: int
    name: str
    password: str          # 敏感字段

@app.post("/users", response_model=UserOut)
def create_user(user: UserIn):
    return user            # password 会被自动过滤

为什么要用响应模型

• 安全：过滤 password 等敏感字段
• 规范：响应结构固定，前端可依赖
• 文档：Swagger 自动展示响应结构
• 校验：响应数据不符合模型时报错（利于调试）

返回列表

@app.get("/users", response_model=list[UserOut])
def get_users():
    return [
        {"id": 1, "name": "张三", "password": "x"},
        {"id": 2, "name": "李四", "password": "y"},
    ]
    # 返回时 password 字段被过滤

包含与排除字段

@app.get("/items", response_model=ItemOut)
def get_item():
    item = {"id": 1, "name": "键盘", "price": 299, "secret": "内部"}
    return item
    # 仅返回 ItemOut 中声明的字段

别名响应模型

class ItemPublic(BaseModel):
    id: int
    name: str

class ItemAdmin(ItemPublic):
    price: float
    internal_note: str

普通用户接口用 ItemPublic，管理接口用 ItemAdmin，通过继承复用。

响应状态码

from fastapi import status

@app.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def create_user(user: UserIn):
    return user

设置响应头

@app.get("/items/{id}", response_model=ItemOut)
def get_item(id: int, response: Response):
    response.headers["X-Request-ID"] = "req-123"
    return {"id": id, "name": "键盘"}

FastAPI 响应状态码

为接口设置响应状态码，用 HTTP 语义准确表达操作结果。

自定义状态码

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items", status_code=201)
def create_item():
    return {"id": 1}

# 等价写法
@app.post("/items", status_code=status.HTTP_201_CREATED)
def create_item2():
    return {"id": 1}

常用状态码速查

常量	值	场景
HTTP_200_OK	200	查询成功（默认）
HTTP_201_CREATED	201	创建成功
HTTP_204_NO_CONTENT	204	删除成功（无内容）
HTTP_400_BAD_REQUEST	400	请求错误
HTTP_401_UNAUTHORIZED	401	未认证
HTTP_403_FORBIDDEN	403	无权限
HTTP_404_NOT_FOUND	404	资源不存在
HTTP_422_UNPROCESSABLE_ENTITY	422	校验失败（FastAPI 默认）
HTTP_500_INTERNAL_SERVER_ERROR	500	服务器错误

动态返回状态码

from fastapi import Response

@app.post("/dynamic")
def dynamic(response: Response):
    # 根据业务逻辑动态设置
    if some_condition:
        response.status_code = 200
        return {"ok": True}
    response.status_code = 202
    return {"accepted": True}

错误状态码

from fastapi import HTTPException

@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id > 1000:
        raise HTTPException(status_code=404, detail="商品不存在")
    return {"id": item_id}

FastAPI 文件上传

FastAPI 处理文件上传非常简单，支持单文件、多文件、表单混合。

单文件上传

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/upload")
async def upload_file(file: UploadFile):
    content = await file.read()          # 读取内容
    return {
        "filename": file.filename,
        "size": len(content),
        "content_type": file.content_type,
    }

UploadFile vs bytes

方式	特点
UploadFile	流式读取，适合大文件，有 filename/content_type 元信息
bytes	整体读入内存，适合小文件

@app.post("/upload-small")
async def upload_small(file: bytes = File()):
    return {"size": len(file)}

保存文件到磁盘

import shutil
from pathlib import Path

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/upload-save")
async def upload_and_save(file: UploadFile):
    # 安全处理文件名
    safe_name = Path(file.filename).name
    dest = UPLOAD_DIR / safe_name

    # 流式写入
    with dest.open("wb") as f:
        shutil.copyfileobj(file.file, f)

    return {"saved": str(dest), "size": dest.stat().st_size}

多文件上传

@app.post("/upload-multiple")
async def upload_multiple(files: list[UploadFile]):
    results = []
    for f in files:
        content = await f.read()
        results.append({"name": f.filename, "size": len(content)})
    return {"files": results}

文件 + 表单字段

from fastapi import Form

@app.post("/post-with-image")
async def create_post(
    title: str = Form(),
    content: str = Form(),
    cover: UploadFile | None = File(None),   # 可选文件
):
    return {"title": title, "cover": cover.filename if cover else None}

文件类型校验

ALLOWED = {"image/png", "image/jpeg", "image/gif"}

@app.post("/upload-image")
async def upload_image(file: UploadFile):
    if file.content_type not in ALLOWED:
        raise HTTPException(status_code=400, detail="仅支持 PNG/JPEG/GIF")
    return {"ok": True}

FastAPI 请求头与 Cookie

读取请求头和 Cookie，实现语言切换、认证信息获取等功能。

读取请求头

from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/headers")
def read_headers(
    user_agent: str | None = Header(default=None),
    accept_language: str | None = Header(default=None),
):
    return {
        "user_agent": user_agent,
        "accept_language": accept_language,
    }

读取认证头

@app.get("/auth-info")
def auth_info(authorization: str | None = Header(default=None)):
    if authorization and authorization.startswith("Bearer "):
        token = authorization.removeprefix("Bearer ")
        return {"token": token}
    return {"token": None}

读取 Cookie

from fastapi import Cookie

@app.get("/cookies")
def read_cookies(session: str | None = Cookie(default=None)):
    return {"session": session}

写入 Cookie

from fastapi import Response

@app.post("/set-cookie")
def set_cookie(response: Response):
    response.set_cookie(
        key="session_id",
        value="abc123",
        httponly=True,       # JS 不可读（防 XSS）
        max_age=3600,        # 1 小时
        path="/",
    )
    return {"ok": True}

完整示例：记住登录状态

from fastapi import FastAPI, Response, Cookie

app = FastAPI()

@app.post("/login")
def login(username: str, response: Response):
    # 登录成功，写 Cookie
    response.set_cookie("user", username, httponly=True, max_age=86400)
    return {"ok": True}

@app.get("/whoami")
def whoami(user: str | None = Cookie(default=None)):
    if not user:
        return {"user": None}
    return {"user": user}

请求头命名规则

FastAPI 自动把 user-agent 映射到参数 user_agent（连字符转下划线）；想保持原样可传 alias：

@app.get("/x")
def read_x(x_request_id: str | None = Header(default=None, alias="X-Request-ID")):
    return {"id": x_request_id}

FastAPI 错误处理

统一的错误处理让接口在出错时返回清晰、一致的信息。

HTTPException 基础

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id > 1000:
        raise HTTPException(status_code=404, detail="商品不存在")
    return {"id": item_id}

// 响应 404：
{"detail": "商品不存在"}

自定义响应头

@app.get("/protected")
def protected():
    raise HTTPException(
        status_code=401,
        detail="请先登录",
        headers={"WWW-Authenticate": "Bearer"},
    )

统一错误响应格式

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class BizError(HTTPException):
    def __init__(self, code: int, message: str):
        super().__init__(status_code=code, detail=message)

@app.exception_handler(BizError)
async def biz_error_handler(request: Request, exc: BizError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"code": exc.status_code, "message": exc.detail, "data": None},
    )

@app.get("/fail")
def fail():
    raise BizError(404, "资源不存在")
# → {"code": 404, "message": "资源不存在", "data": null}

全局异常处理

import logging

logger = logging.getLogger("app")

@app.exception_handler(Exception)
async def global_error(request: Request, exc: Exception):
    logger.exception("unhandled error")   # 完整堆栈进日志
    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": "服务器内部错误", "data": None},
    )

校验错误自定义

from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"code": 422, "message": "参数校验失败", "errors": exc.errors()},
    )

错误处理最佳实践

• 业务错误用自定义异常 + 全局处理器，统一格式
• 不要把内部堆栈返回给客户端（信息泄露）
• 日志记录完整错误，响应只给友好信息
• 按错误类型选择状态码（400/401/403/404/409/500）

FastAPI 中间件

中间件（Middleware）是处理请求和响应的"钩子"，在路由执行之前和之后运行，适合日志、计时、鉴权、异常兜底等横切逻辑。

中间件的执行时机

请求进入
  ↓
中间件（请求处理前）  ← 可记录日志、校验、拦截
  ↓
路由处理
  ↓
中间件（响应处理后）  ← 可加响应头、统计耗时
  ↓
响应返回客户端

基础中间件

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start = time.time()

    # 请求进入路由前
    response = await call_next(request)   # 调用下一个处理

    # 响应返回前
    process_time = time.time() - start
    response.headers["X-Process-Time"] = str(process_time)
    return response

日志中间件

import logging

logger = logging.getLogger("api")

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"→ {request.method} {request.url.path}")
    response = await call_next(request)
    logger.info(f"← {request.method} {request.url.path} {response.status_code}")
    return response

请求拦截中间件

from fastapi.responses import JSONResponse

BLOCKED_IPS = {"192.168.1.100"}

@app.middleware("http")
async def block_ips(request: Request, call_next):
    client_ip = request.client.host if request.client else "unknown"
    if client_ip in BLOCKED_IPS:
        return JSONResponse(status_code=403, content={"detail": "IP 已被封禁"})
    return await call_next(request)

认证中间件

@app.middleware("http")
async def auth_check(request: Request, call_next):
    # 白名单路径跳过
    if request.url.path in ("/api/login", "/docs", "/openapi.json"):
        return await call_next(request)

    token = request.headers.get("Authorization", "")
    if not token.startswith("Bearer "):
        return JSONResponse(status_code=401, content={"detail": "未认证"})
    return await call_next(request)

使用内置中间件

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["www.example.com", "*.example.com"],
)

中间件 vs 依赖注入

对比项	中间件	依赖注入
作用范围	所有请求	指定端点
拦截能力	可在路由前直接返回	可以（抛异常）
获取参数	Request 对象	函数参数
典型用途	日志、计时、IP 拦截	数据库、认证、分页

FastAPI CORS 跨域

CORS（跨域资源共享）控制浏览器能否跨域访问你的接口。前后端分离部署时几乎必配。

什么是跨域

浏览器同源策略默认禁止网页访问不同源（协议/域名/端口任一不同）的接口。例如前端在 http://localhost:3000，API 在 http://api.example.com，就属于跨域。

启用 CORS

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://www.example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

参数说明

参数	说明
allow_origins	允许的来源列表（或 ["*"] 全部）
allow_credentials	是否允许携带 Cookie（与 * 不能同时用）
allow_methods	允许的方法（GET/POST/PUT/DELETE）
allow_headers	允许的请求头
expose_headers	允许前端读取的响应头
max_age	预检请求缓存时间（秒）

生产配置建议

# 开发环境：全部放开
app.add_middleware(CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 生产环境：只允许指定域名
ALLOWED_ORIGINS = [
    "https://www.laowang.click",
    "https://admin.laowang.click",
]

app.add_middleware(CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Content-Type", "Authorization"],
)

CORS 预检请求

非简单请求（带自定义头、非 GET/POST）会先发 OPTIONS 预检，服务器返回允许策略后，浏览器才发真实请求。FastAPI 的 CORSMiddleware 自动处理预检。

常见报错

报错	原因	解决
No 'Access-Control-Allow-Origin'	未配置 CORS	添加 CORSMiddleware
allow_credentials 与 * 冲突	携带 Cookie 时不能全放	改为明确域名列表
预检 405	OPTIONS 未处理	确认中间件已加载

FastAPI 静态文件

让 FastAPI托管静态文件（图片、CSS、JS、上传文件等），前后端同域部署。

挂载静态目录

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# 将 ./static 目录映射到 /static 路径
app.mount("/static", StaticFiles(directory="static"), name="static")

访问 http://127.0.0.1:8000/static/logo.png 即可获取 static/logo.png。

项目结构

project/
├── main.py
├── static/
│   ├── css/style.css
│   ├── js/app.js
│   └── images/logo.png
└── uploads/          # 用户上传文件

托管上传文件

import shutil
from pathlib import Path
from fastapi import FastAPI, UploadFile

app = FastAPI()

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

# 上传文件目录也可静态访问
app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads")

@app.post("/upload")
async def upload(file: UploadFile):
    dest = UPLOAD_DIR / file.filename
    with dest.open("wb") as f:
        shutil.copyfileobj(file.file, f)
    # 返回可访问的 URL
    return {"url": f"/uploads/{file.filename}"}

返回 HTML 页面

from fastapi.responses import HTMLResponse, FileResponse

@app.get("/", response_class=HTMLResponse)
async def home():
    return FileResponse("index.html")   # 返回静态 HTML

# 或直接返回 HTML 字符串
@app.get("/hello", response_class=HTMLResponse)
async def hello():
    return "<h1>Hello</h1><p>静态页面</p>"

前后端同域部署

把前端构建产物放入 static（或指定目录），FastAPI 同时提供页面与 API：

# 前端 dist 目录
app.mount("/", StaticFiles(directory="dist", html=True), name="frontend")

# 注意：/api 路由要定义在 mount "/" 之前

FastAPI 安全认证

为 API 添加安全认证，保护敏感接口。本节介绍最常用的 JWT 方案。

认证方式概览

方案	特点	适用
JWT Token	无状态、标准	前后端分离（推荐）
OAuth2	授权协议	第三方登录
API Key	简单	服务间调用
Session	服务端存储	传统 Web

安装依赖

pip install "python-jose[cryptography]" "passlib[bcrypt]" python-multipart

密码哈希

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

生成与验证 JWT

from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt

SECRET_KEY = "change-this-secret-key-in-production"
ALGORITHM = "HS256"
TOKEN_EXPIRE_MINUTES = 60

def create_token(user_id: int) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=TOKEN_EXPIRE_MINUTES)
    payload = {"sub": str(user_id), "exp": expire}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str) -> int:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return int(payload["sub"])
    except JWTError:
        return None

登录接口

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class LoginRequest(BaseModel):
    username: str
    password: str

@app.post("/api/login")
def login(req: LoginRequest):
    # 实际应从数据库校验用户
    if req.username != "admin" or not verify_password(req.password, stored_hash):
        raise HTTPException(status_code=401, detail="用户名或密码错误")
    token = create_token(user_id=1)
    return {"token": token, "token_type": "bearer"}

保护接口（依赖注入）

from fastapi import Depends, Header

def get_current_user(authorization: str = Header(default="")):
    token = authorization.removeprefix("Bearer ")
    user_id = decode_token(token)
    if user_id is None:
        raise HTTPException(status_code=401, detail="Token 无效或已过期")
    return user_id

@app.get("/api/me")
def read_me(user_id: int = Depends(get_current_user)):
    return {"user_id": user_id}

# 未带 Token 访问 → 401

安全要点

• SECRET_KEY 放入环境变量，绝不写死代码里
• 密码用 bcrypt 等加盐哈希存储，绝不存明文
• Token 设置有效期（exp），过期必须重新登录
• 全部接口启用 HTTPS 传输

FastAPI 测试

为 FastAPI 编写自动化测试，保证接口质量，防止回归。

安装测试工具

pip install pytest httpx

基础测试

# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_root():
    resp = client.get("/")
    assert resp.status_code == 200
    assert resp.json() == {"message": "Hello, FastAPI!"}

# 运行测试
pytest -v

测试 CRUD 接口

def test_create_and_get_item():
    # 创建
    resp = client.post("/items", json={"name": "键盘", "price": 299})
    assert resp.status_code == 201
    item_id = resp.json()["id"]

    # 查询
    resp = client.get(f"/items/{item_id}")
    assert resp.status_code == 200
    assert resp.json()["name"] == "键盘"

    # 更新
    resp = client.put(f"/items/{item_id}", json={"name": "鼠标", "price": 99})
    assert resp.status_code == 200

    # 删除
    resp = client.delete(f"/items/{item_id}")
    assert resp.status_code == 204

测试错误场景

def test_item_not_found():
    resp = client.get("/items/99999")
    assert resp.status_code == 404

def test_validation_error():
    # price 缺省 → 422
    resp = client.post("/items", json={"name": "缺价格"})
    assert resp.status_code == 422

测试认证

def test_auth_required():
    # 未带 Token → 401
    resp = client.get("/api/me")
    assert resp.status_code == 401

    # 带 Token → 200
    login = client.post("/api/login", json={"username": "admin", "password": "secret"})
    token = login.json()["token"]
    resp = client.get("/api/me", headers={"Authorization": f"Bearer {token}"})
    assert resp.status_code == 200

测试数据库（fixture）

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture
def db():
    # 使用临时数据库
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    yield Session()
    engine.dispose()

测试约定

• 测试文件命名 test_*.py
• 一个接口一个测试函数，场景分开
• 覆盖成功、失败、边界三种情况
• 测试用独立数据库，不污染生产数据

FastAPI 数据库集成

将 FastAPI 与数据库集成，实现数据持久化。本节使用 SQLAlchemy ORM + MySQL。

安装依赖

pip install sqlalchemy pymysql

数据库连接（database.py）

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DB_URL = "mysql+pymysql://user:password@127.0.0.1:3306/mydb?charset=utf8mb4"

engine = create_engine(DB_URL, pool_pre_ping=True, pool_recycle=3600)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

class Base(DeclarativeBase):
    pass

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

定义模型（models.py）

from sqlalchemy import String, Integer, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(100), default="")
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

创建表

# 在 main.py 中或初始化脚本中
from .models import Base
from .database import engine

Base.metadata.create_all(bind=engine)   # 开发用；生产建议 Alembic 迁移

CRUD 操作

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .database import get_db
from .models import User

app = FastAPI()

@app.post("/api/users")
def create_user(username: str, db: Session = Depends(get_db)):
    # 查重
    if db.query(User).filter(User.username == username).first():
        raise HTTPException(status_code=409, detail="用户名已存在")
    user = User(username=username)
    db.add(user)
    db.commit()          # 提交事务
    db.refresh(user)     # 刷新获取 id
    return {"id": user.id, "username": user.username}

@app.get("/api/users")
def list_users(db: Session = Depends(get_db)):
    return [
        {"id": u.id, "username": u.username}
        for u in db.query(User).order_by(User.id.desc()).all()
    ]

@app.get("/api/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return {"id": user.id, "username": user.username}

事务管理

def create_with_commit(db, user):
    try:
        db.add(user)
        db.commit()
    except Exception:
        db.rollback()     # 出错回滚
        raise

数据库连接池配置

engine = create_engine(
    DB_URL,
    pool_size=10,          # 池大小
    max_overflow=20,       # 最大溢出连接
    pool_pre_ping=True,    # 连接前检查（防断连）
    pool_recycle=3600,     # 1 小时回收
    echo=False,            # True 打印 SQL（调试用）
)

FastAPI Web 应用

本节把前面所有知识串联起来，构建一个完整的 FastAPI Web 应用。

项目结构

myblog/
├── app/
│   ├── __init__.py
│   ├── main.py          # 入口：创建 app、注册路由
│   ├── database.py      # 数据库连接
│   ├── models.py        # ORM 模型
│   ├── schemas.py       # Pydantic 模型
│   └── routers/
│       ├── __init__.py
│       ├── auth.py      # 认证路由
│       ├── posts.py     # 文章路由
│       └── users.py     # 用户路由
├── static/              # 静态文件
├── templates/           # 模板（如用 Jinja2）
└── requirements.txt

模块化路由

# app/routers/posts.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from ..database import get_db
from .. import models, schemas

router = APIRouter(prefix="/api/posts", tags=["文章"])

@router.get("")
def list_posts(db: Session = Depends(get_db)):
    return db.query(models.Post).all()

@router.post("", response_model=schemas.PostOut, status_code=201)
def create_post(post: schemas.PostIn, db: Session = Depends(get_db)):
    obj = models.Post(**post.model_dump())
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj

# app/main.py
from fastapi import FastAPI
from .routers import auth, posts, users

app = FastAPI(title="MyBlog API")

# 注册子路由
app.include_router(posts.router)
app.include_router(users.router)
app.include_router(auth.router)

配置管理（环境变量）

import os
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = os.getenv("DB_URL", "sqlite:///./app.db")
    secret_key: str = os.getenv("SECRET_KEY", "dev-secret")
    admin_password: str = os.getenv("ADMIN_PASSWORD", "admin123")

    class Config:
        env_file = ".env"

settings = Settings()

启动命令

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

完整应用示例（教程站模式）

# CodeNest 本站就是 FastAPI 应用：
# - 前端：HTML/CSS/JS（index.html 等）
# - 后端：FastAPI 提供 /api 接口
# - 数据库：MySQL knowledge_site（categories/tutorials/sections 三表）
# - 管理后台：admin.html + /api/admin 接口

Jinja2 模板

Jinja2 是 Python 最流行的模板引擎，FastAPI 通过 Starlette 集成，用于服务端渲染 HTML 页面。

安装

pip install jinja2

配置模板

from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates

app = FastAPI()
templates = Jinja2Templates(directory="templates")

@app.get("/")
def home(request: Request):
    return templates.TemplateResponse(
        request=request,
        name="index.html",
        context={"title": "首页", "items": ["A", "B", "C"]},
    )

模板文件 templates/index.html

<!DOCTYPE html>
<html>
<head><title>{{ title }}</title></head>
<body>
  <h1>{{ title }}</h1>
  <ul>
    {% for item in items %}
      <li>{{ item }}</li>
    {% endfor %}
  </ul>
</body>
</html>

模板语法速查

{{ variable }}           <!-- 输出变量 -->
{% if condition %} ... {% endif %}
{% for item in items %} ... {% endfor %}
{{ item | upper }}        <!-- 过滤器 -->
{% extends "base.html" %} <!-- 继承 -->
{% block content %}{% endblock %}

模板继承

<!-- base.html -->
<html><body>
  <header>公共导航</header>
  {% block content %}{% endblock %}
</body></html>

<!-- page.html -->
{% extends "base.html" %}
{% block content %}
  <p>页面内容</p>
{% endblock %}

静态资源

from fastapi.staticfiles import StaticFiles

app.mount("/static", StaticFiles(directory="static"), name="static")

<link rel="stylesheet" href="/static/css/style.css">

什么时候用 Jinja2

• 服务端渲染的简单页面
• 需要 SEO（搜索引擎收录）的页面
• 快速原型

前后端分离项目（SPA）则用 JSON API + 前端框架，不涉及模板。

SQLAlchemy 模型

定义 SQLAlchemy 模型，把 Python 类映射为数据库表，是 ORM 开发的核心。

模型定义基础

from sqlalchemy import String, Integer, Text, DateTime, ForeignKey, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Category(Base):
    __tablename__ = "categories"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    icon: Mapped[str] = mapped_column(String(20), default="📁")
    sort_order: Mapped[int] = mapped_column(Integer, default=0)

    # 一对多关系
    tutorials: Mapped[list["Tutorial"]] = relationship(back_populates="category")

多表关系

class Tutorial(Base):
    __tablename__ = "tutorials"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    cat_id: Mapped[int] = mapped_column(ForeignKey("categories.id"), nullable=False)
    title: Mapped[str] = mapped_column(String(100))
    slug: Mapped[str] = mapped_column(String(60), default="", index=True)
    desc: Mapped[str] = mapped_column(String(255), default="")
    read_time: Mapped[str] = mapped_column(String(20), default="")
    sort_order: Mapped[int] = mapped_column(Integer, default=0)

    category: Mapped[Category] = relationship(back_populates="tutorials")
    sections: Mapped[list["Section"]] = relationship(
        back_populates="tutorial", cascade="all, delete-orphan",
        order_by="Section.sort_order",
    )

class Section(Base):
    __tablename__ = "sections"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    tutorial_id: Mapped[int] = mapped_column(ForeignKey("tutorials.id"), nullable=False)
    title: Mapped[str] = mapped_column(String(100))
    content: Mapped[str] = mapped_column(Text, default="")
    sort_order: Mapped[int] = mapped_column(Integer, default=0)

    tutorial: Mapped[Tutorial] = relationship(back_populates="sections")

常见列类型

类型	用途
String(n)	变长字符串
Text	长文本
Integer	整数
Float	浮点数
Boolean	布尔
DateTime	日期时间
JSON	JSON 数据

字段约束

id: Mapped[int] = mapped_column(Integer, primary_key=True)        # 主键
name: Mapped[str] = mapped_column(String(50), unique=True)        # 唯一
email: Mapped[str] = mapped_column(String(100), index=True)       # 索引
desc: Mapped[str] = mapped_column(String(255), default="")        # 默认值
cat_id: Mapped[int] = mapped_column(ForeignKey("categories.id"))  # 外键
deleted: Mapped[bool] = mapped_column(Boolean, default=False)     # 软删除标记

创建表

from .database import engine
from . import models

# 开发环境直接建表；生产用 Alembic 迁移
models.Base.metadata.create_all(bind=engine)

Pydantic Schema

Pydantic Schema 定义 API 的输入输出数据结构，与 ORM 模型职责分离。

Schema 与 Model 的区别

对比项	ORM Model	Pydantic Schema
职责	数据库表映射	API 数据校验
位置	models.py	schemas.py
校验	数据库层	应用层（FastAPI 自动）

基础 Schema

from pydantic import BaseModel, Field

# 输入模型：创建教程
class TutorialCreate(BaseModel):
    cat_id: int
    title: str = Field(min_length=1, max_length=100)
    icon: str = "📘"
    accent: str = "#6366f1"
    desc: str = ""
    tags: str = ""
    read_time: str = ""

# 输出模型：返回教程
class TutorialOut(BaseModel):
    id: int
    cat_id: int
    title: str
    desc: str
    sort_order: int

    class Config:
        from_attributes = True   # 允许从 ORM 对象转换

ORM 转 Schema

@app.get("/api/tutorials/{tid}", response_model=TutorialOut)
def get_tutorial(tid: int, db: Session = Depends(get_db)):
    tutorial = db.get(Tutorial, tid)
    if not tutorial:
        raise HTTPException(status_code=404, detail="教程不存在")
    return tutorial   # ORM 对象 → Pydantic 自动转换

嵌套输出

class SectionOut(BaseModel):
    id: int
    title: str
    content: str

class TutorialDetail(TutorialOut):
    sections: list[SectionOut] = []

复合 Schema 用法

class CategoryWithCount(BaseModel):
    id: int
    name: str
    tutorial_count: int   # 计算字段

@app.get("/api/categories")
def get_categories(db: Session = Depends(get_db)):
    rows = db.query(Category, func.count(Tutorial.id)).outerjoin(Tutorial)\
        .group_by(Category.id).all()
    return [CategoryWithCount(id=c.id, name=c.name, tutorial_count=n) for c, n in rows]

Schema 最佳实践

• 输入/输出模型分离（Create/Update/Out）
• 输出模型不放敏感字段
• 嵌套结构用多个 Schema 组合
• from_attributes = True 支持 ORM 对象直接返回

路由

用 APIRouter 组织路由，把相关接口分组到独立文件，代码更整洁。

为什么拆路由

• 文件职责清晰（用户、文章、管理分开）
• 避免 main.py 无限膨胀
• 可复用前缀和标签
• 团队并行开发互不干扰

创建子路由

# app/routers/tutorials.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from ..database import get_db
from .. import models, schemas

# 前缀 + 标签：所有路由统一 /api/tutorials，文档归"教程"组
router = APIRouter(prefix="/api/tutorials", tags=["教程"])

@router.get("", response_model=list[schemas.TutorialOut])
def list_tutorials(db: Session = Depends(get_db)):
    return db.query(models.Tutorial).order_by(models.Tutorial.sort_order).all()

@router.get("/{tid}", response_model=schemas.TutorialDetail)
def get_tutorial(tid: int, db: Session = Depends(get_db)):
    tutorial = db.get(models.Tutorial, tid)
    if not tutorial:
        raise HTTPException(status_code=404, detail="教程不存在")
    return tutorial

@router.post("", response_model=schemas.TutorialOut, status_code=201)
def create_tutorial(data: schemas.TutorialCreate, db: Session = Depends(get_db)):
    obj = models.Tutorial(**data.model_dump())
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj

注册路由

# app/main.py
from fastapi import FastAPI
from .routers import tutorials, sections, admin

app = FastAPI(title="CodeNest API")

app.include_router(tutorials.router)
app.include_router(sections.router)
app.include_router(admin.router)

子路由带认证

from fastapi import Depends
from ..auth import get_current_user

# 整个路由器都要求认证
router = APIRouter(
    prefix="/api/admin",
    tags=["管理"],
    dependencies=[Depends(get_current_user)],   # 统一依赖
)

响应模型与状态码

@router.delete("/tutorials/{tid}", status_code=204)
def delete_tutorial(tid: int, db: Session = Depends(get_db)):
    obj = db.get(models.Tutorial, tid)
    if not obj:
        raise HTTPException(status_code=404, detail="教程不存在")
    db.delete(obj)
    db.commit()

Alembic 数据库迁移

Alembic 是 SQLAlchemy 官方数据库迁移工具，管理表结构变更，生产环境必备。

为什么需要迁移

• 开发改模型后，数据库表结构需要同步升级
• 多人协作时结构变更可追溯
• 上线前可预览/回滚变更

安装与初始化

pip install alembic

# 初始化（生成 alembic/ 目录和 alembic.ini）
alembic init alembic

配置连接

编辑 alembic/env.py，指定数据库 URL 和目标模型：

from app.database import Base
from app import models  # 导入所有模型

target_metadata = Base.metadata

# 或直接在 env.py 中配置 URL
# config.set_main_option("sqlalchemy.url", "mysql+pymysql://user:pass@localhost/mydb")

生成迁移脚本

# 自动对比模型与数据库，生成迁移脚本
alembic revision --autogenerate -m "add tutorial table"

应用迁移

# 升级到最新版本
alembic upgrade head

# 回退一个版本
alembic downgrade -1

# 查看当前版本
alembic current

# 查看历史
alembic history

修改模型后流程

1. 修改 models.py
2. alembic revision --autogenerate -m "描述变更"
3. 检查生成的迁移脚本（确认无误）
4. alembic upgrade head
5. 测试接口

手动编写迁移

# alembic/versions/xxxx_add_column.py
def upgrade():
    op.add_column("tutorials", sa.Column("cover", sa.String(255), nullable=True))

def downgrade():
    op.drop_column("tutorials", "cover")

最佳实践

• 每次模型变更都生成迁移，别手动改库
• 上线前先 alembic upgrade head 预览
• 迁移脚本提交到版本控制
• 生产库变更前先备份

关键词搜索与分页

实现关键词搜索与分页，这是内容站 API 的高频需求。

分页查询

from fastapi import FastAPI, Depends, Query
from sqlalchemy.orm import Session

@app.get("/api/tutorials")
def list_tutorials(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    db: Session = Depends(get_db),
):
    # 总数
    total = db.query(models.Tutorial).count()

    # 分页查询
    items = (
        db.query(models.Tutorial)
        .order_by(models.Tutorial.sort_order)
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    return {
        "items": items,
        "pagination": {
            "page": page,
            "page_size": page_size,
            "total": total,
            "total_pages": (total + page_size - 1) // page_size,
        },
    }

关键词搜索

@app.get("/api/search")
def search(
    q: str = Query(min_length=1),
    db: Session = Depends(get_db),
):
    # LIKE 模糊匹配标题
    items = (
        db.query(models.Tutorial)
        .filter(models.Tutorial.title.like(f"%{q}%"))
        .order_by(models.Tutorial.sort_order)
        .all()
    )
    return {"items": items, "keyword": q}

多字段搜索

from sqlalchemy import or_

@app.get("/api/search")
def search(q: str, db: Session = Depends(get_db)):
    kw = f"%{q}%"
    items = db.query(models.Tutorial).filter(or_(
        models.Tutorial.title.like(kw),
        models.Tutorial.desc.like(kw),
        models.Tutorial.tags.like(kw),
    )).all()
    return {"items": items}

搜索 + 分类过滤 + 分页

@app.get("/api/tutorials")
def list_tutorials(
    page: int = 1,
    page_size: int = 10,
    q: str | None = None,
    cat_id: int | None = None,
    db: Session = Depends(get_db),
):
    query = db.query(models.Tutorial)

    # 组合过滤条件
    if q:
        query = query.filter(models.Tutorial.title.like(f"%{q}%"))
    if cat_id:
        query = query.filter(models.Tutorial.cat_id == cat_id)

    total = query.count()
    items = query.order_by(models.Tutorial.sort_order)\
        .offset((page - 1) * page_size).limit(page_size).all()

    return {"items": items, "total": total}

SQL 注入安全

SQLAlchemy 的 like() 方法自动参数化，安全防注入。即使把用户输入拼进 f"%{q}%"，值也是作为参数传递，不会被解析为 SQL。

用户系统

构建完整的用户系统：注册、登录、鉴权、个人资料。

用户模型

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    password_hash: Mapped[str] = mapped_column(String(255))
    email: Mapped[str] = mapped_column(String(100), default="")
    avatar: Mapped[str] = mapped_column(String(255), default="")
    is_admin: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

注册接口

from passlib.context import CryptContext
from pydantic import BaseModel, Field

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class RegisterIn(BaseModel):
    username: str = Field(min_length=3, max_length=20, pattern=r"^\w+$")
    password: str = Field(min_length=6, max_length=50)
    email: str = ""

@app.post("/api/register", status_code=201)
def register(data: RegisterIn, db: Session = Depends(get_db)):
    # 查重
    if db.query(User).filter(User.username == data.username).first():
        raise HTTPException(status_code=409, detail="用户名已存在")

    user = User(
        username=data.username,
        password_hash=pwd_context.hash(data.password),   # 哈希存储
        email=data.email,
    )
    db.add(user)
    db.commit()
    return {"id": user.id, "username": user.username}

登录接口

class LoginIn(BaseModel):
    username: str
    password: str

@app.post("/api/login")
def login(data: LoginIn, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.username == data.username).first()
    if not user or not pwd_context.verify(data.password, user.password_hash):
        raise HTTPException(status_code=401, detail="用户名或密码错误")
    token = create_token(user.id)
    return {"token": token, "username": user.username, "is_admin": user.is_admin}

获取当前用户

def get_current_user(
    authorization: str = Header(default=""),
    db: Session = Depends(get_db),
) -> User:
    token = authorization.removeprefix("Bearer ")
    user_id = decode_token(token)
    if user_id is None:
        raise HTTPException(status_code=401, detail="请先登录")
    user = db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    return user

@app.get("/api/me")
def read_me(user: User = Depends(get_current_user)):
    return {"id": user.id, "username": user.username, "email": user.email}

管理员权限

def require_admin(user: User = Depends(get_current_user)):
    if not user.is_admin:
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return user

@app.get("/api/admin/stats")
def admin_stats(user: User = Depends(require_admin)):
    return {"users": db.query(User).count()}

安全要点

• 密码必须哈希存储（bcrypt），绝不存明文
• 登录接口做限流，防暴力破解
• Token 有效期适中（如 1-7 天）
• 管理接口独立鉴权（is_admin）

收藏功能与后台管理

实现收藏功能与后台管理，让教程站具备交互与管理能力。

收藏功能设计

class Favorite(Base):
    __tablename__ = "favorites"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
    tutorial_id: Mapped[int] = mapped_column(ForeignKey("tutorials.id"), index=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    __table_args__ = (UniqueConstraint("user_id", "tutorial_id"),)  # 唯一：防重复收藏

收藏接口

@app.post("/api/favorites/{tutorial_id}", status_code=201)
def add_favorite(
    tutorial_id: int,
    user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    exists = db.query(Favorite).filter_by(
        user_id=user.id, tutorial_id=tutorial_id
    ).first()
    if exists:
        return {"ok": True, "message": "已收藏过"}

    db.add(Favorite(user_id=user.id, tutorial_id=tutorial_id))
    db.commit()
    return {"ok": True}

@app.delete("/api/favorites/{tutorial_id}", status_code=204)
def remove_favorite(
    tutorial_id: int,
    user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    fav = db.query(Favorite).filter_by(
        user_id=user.id, tutorial_id=tutorial_id
    ).first()
    if fav:
        db.delete(fav)
        db.commit()

收藏列表

@app.get("/api/favorites")
def list_favorites(
    user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    rows = (
        db.query(Tutorial)
        .join(Favorite, Favorite.tutorial_id == Tutorial.id)
        .filter(Favorite.user_id == user.id)
        .all()
    )
    return {"items": rows}

后台管理（管理员）

# 管理路由（统一要求管理员权限）
router = APIRouter(prefix="/api/admin", tags=["管理"],
                   dependencies=[Depends(require_admin)])

@router.get("/tutorials")
def admin_list_tutorials(db: Session = Depends(get_db)):
    return db.query(Tutorial).order_by(Tutorial.id).all()

@router.put("/tutorials/{tid}")
def admin_update_tutorial(tid: int, data: schemas.TutorialIn,
                          db: Session = Depends(get_db)):
    obj = db.get(Tutorial, tid)
    if not obj:
        raise HTTPException(status_code=404, detail="教程不存在")
    for k, v in data.model_dump().items():
        setattr(obj, k, v)
    db.commit()
    return obj

@router.put("/sections/{sid}")
def admin_update_section(sid: int, content: str, db: Session = Depends(get_db)):
    section = db.get(Section, sid)
    if not section:
        raise HTTPException(status_code=404, detail="章节不存在")
    section.content = content
    db.commit()
    return section

后台功能清单

• 教程/章节 CRUD
• 分类管理
• 内容编辑（本站就是通过 PUT /api/admin/sections 更新章节）
• 统计（用户数、收藏数）

打包上线

把 FastAPI 应用打包上线，部署到生产服务器（Nginx + uvicorn 或 Gunicorn）。

上线前检查清单

• 关闭 --reload（调试模式）
• SECRET_KEY / ADMIN_PASSWORD 等放入环境变量
• 数据库使用独立生产库与账号
• HTTPS 证书已配置
• 日志已开启（输出到文件）
• 依赖锁定（requirements.txt）

生产启动方式

# 方式一：uvicorn 单进程
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

# 方式二：Gunicorn + uvicorn worker（推荐）
pip install gunicorn
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

Nginx 反向代理

server {
    listen 80;
    server_name www.laowang.click;

    # 静态资源由 nginx 直接服务（更快）
    location /static/ {
        alias /www/wwwroot/www/static/;
        expires 7d;
    }

    # API 反代到 FastAPI
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

HTTPS 配置

server {
    listen 443 ssl http2;
    server_name www.laowang.click;

    ssl_certificate     /etc/ssl/cert.pem;
    ssl_certificate_key /etc/ssl/key.pem;

    location / {
        proxy_pass http://127.0.0.1:8000;
    }
}

server {
    listen 80;
    server_name www.laowang.click;
    return 301 https://$host$request_uri;   # HTTP 跳转 HTTPS
}

systemd 服务管理

# /etc/systemd/system/myapi.service
[Unit]
Description=FastAPI App
After=network.target

[Service]
WorkingDirectory=/www/wwwroot/myapp
ExecStart=/www/wwwroot/myapp/venv/bin/gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 127.0.0.1:8000
Restart=always
EnvironmentFile=/www/wwwroot/myapp/.env

[Install]
WantedBy=multi-user.target

systemctl daemon-reload
systemctl enable myapi
systemctl start myapi
systemctl status myapi

宝塔面板部署（本站方式）

1. 创建 Python 项目（网站 → Python 项目）
2. 上传代码，安装依赖（pip install -r requirements.txt）
3. 启动命令设为 python -m app.main
4. 打开"外网映射"（自动生成 Nginx 反代）
5. 申请 SSL 证书（Let's Encrypt）

上线后检查

• curl http://127.0.0.1:8000/api/health 健康检查
• 访问域名验证页面与 API
• 查看日志确认无异常