FastAPI 学习文档：https://fastapi.tiangolo.com/zh/learn/

环境准备

本教程统一使用 uv 管理 Python 版本和依赖，uv 的完整用法见使用uv管理环境，这里只列出会用到的几个命令：

命令	作用
`uv init my-project`	初始化项目，生成 `pyproject.toml`
`uv python pin 3.12`	锁定项目使用的 Python 版本
`uv add xxx`	添加依赖，并更新 `uv.lock`
`uv add --dev xxx`	添加开发依赖（测试、迁移工具等）
`uv sync`	按锁文件同步环境
`uv run xxx`	在项目虚拟环境中执行命令，无需手动激活

快速开始

1. 初始化项目

1 2	uv init my-fastapi-project cd my-fastapi-project

2. 安装 FastAPI 和服务器

1	uv add fastapi uvicorn

也可以直接安装 uv add "fastapi[standard]"，它会一并装上 uvicorn、python-multipart、httpx 等常用依赖，以及 fastapi 命令行工具。

3. 编写 main.py

from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World from uv!"}

if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

4. 运行项目

以下三种方式任选其一：

# 方式一：uvicorn 命令行，--reload 表示代码修改后自动重启
uv run uvicorn main:app --reload

# 方式二：直接运行脚本，走 if __name__ == "__main__" 分支
uv run python main.py

# 方式三：安装了 fastapi[standard] 后可用
uv run fastapi dev main.py

main:app 的含义是「main.py 模块中名为 app 的对象」。

启动后 FastAPI 会自动生成两套交互式接口文档：

Swagger UI：http://127.0.0.1:8000/docs

ReDoc：http://127.0.0.1:8000/redoc

原始 OpenAPI 描述：http://127.0.0.1:8000/openapi.json

`async def` 还是 `def`？

FastAPI 两种写法都支持，区别在于执行方式：

async def：在事件循环中直接执行。函数内部只能调用异步的 IO 操作（await 异步数据库驱动、httpx.AsyncClient 等），一旦调用了 time.sleep()、requests.get() 这类阻塞操作，整个服务都会被卡住。
def：FastAPI 会把它丢到线程池中执行，不会阻塞事件循环，适合调用同步库。

简单的判断原则：用到的库支持 await 就写 async def，否则写 def。

路径操作

方法装饰器

每个标准 HTTP 方法都有对应的装饰器：

@app.get()：获取数据，不改变服务器状态。
@app.post()：创建新数据。
@app.put()：完整更新已有资源。
@app.patch()：部分更新已有资源。
@app.delete()：删除资源。

另外还有 @app.options()、@app.head() 等。

from fastapi import FastAPI
app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello World from uv!"}

@app.get("/getMethod")
async def get():
    return {"method": "GET"}

@app.post("/postMethod")
async def post():
    return {"method": "POST"}

@app.delete("/deleteMethod")
async def delete():
    return {"method": "DELETE"}

路径装饰器常用参数

from fastapi import FastAPI
app = FastAPI()

@app.get("/test",
         tags=["这是一个测试接口"],                          # 文档中的分组
         summary="这是测试接口的 summary",                   # 接口简述
         description="这是测试接口的 description",           # 接口详细描述
         response_description="这是测试接口的 response description",
         deprecated=False,                                   # 是否标记为废弃
)
async def root():
    return {"message": "测试数据"}

路由分发

调整项目结构为

my-fastapi-project/
├── apps/
│   ├── app01/
│   │   ├── __init__.py
│   │   └── urls.py
│   ├── app02/
│   │   ├── __init__.py
│   │   └── urls.py
│   └── __init__.py
├── main.py

app01/urls.py

from fastapi import APIRouter

app01_router = APIRouter()

@app01_router.get("/test")
async def test():
    return {"message": "app01测试数据"}

app02/urls.py

from fastapi import APIRouter

app02_router = APIRouter()

@app02_router.get("/test")
async def test():
    return {"message": "app02测试数据"}

main.py

import uvicorn
from fastapi import FastAPI
from apps.app01.urls import app01_router as app01_router
from apps.app02.urls import app02_router as app02_router

app = FastAPI()

app.include_router(app01_router, prefix="/app01", tags=["app01"])
app.include_router(app02_router, prefix="/app02", tags=["app02"])

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

也可以在创建 APIRouter 时直接指定公共配置，include_router 时就不用再写了：
1
app01_router = APIRouter(prefix="/app01", tags=["app01"])

请求与响应

路径参数

import uvicorn
from fastapi import FastAPI

app = FastAPI()

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

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

声明了类型 int 后，FastAPI 会自动做类型转换和校验：访问 /user/abc 会直接返回 422，并在响应中说明哪个参数校验失败。

注意路由匹配顺序：FastAPI 按照声明顺序匹配路由。如果同时有 /user/me 和 /user/{user_id}，必须把 /user/me 写在前面，否则 me 会被当作 user_id 去解析，结果返回 422。

枚举限定取值

from enum import Enum

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    # 只能传 alexnet / resnet / lenet，文档中会显示为下拉框
    return {"model_name": model_name, "value": model_name.value}

使用 `Path` 校验

from typing import Annotated
from fastapi import Path

@app.get("/items/{item_id}")
async def get_item(item_id: Annotated[int, Path(title="商品ID", ge=1, le=1000)]):
    return {"item_id": item_id}

常用的数值校验参数：gt（大于）、ge（大于等于）、lt（小于）、le（小于等于）。

查询参数

路径函数中不属于路径参数的参数，会被自动当作查询参数，也就是 URL 中 ? 后面用 & 分隔的 key=value 键值对。

from typing import Union, Optional
import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/user/{category}")
async def get_user(category: str, phone: Union[str, None] = None, name: Optional[str] = None):
    # 有默认参数，就是非必选项
    # Union 表示 phone 可以是 str 类型，也可以是 None 类型
    # Optional 表示 name 可以是 str 类型，也可以是 None 类型，是Union的简化
    return {"category": category, "phone": phone, "name": name}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Python 3.10+ 中，Union[str, None]、Optional[str] 都可以简写成 str | None，后文会混用这几种写法。

使用 `Query` 校验

和 Path 类似，Query 可以给查询参数加上额外的校验与文档信息，推荐配合 Annotated 使用：

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(min_length=2, max_length=50, description="搜索关键词")] = None,
    page: Annotated[int, Query(ge=1)] = 1,
    size: Annotated[int, Query(ge=1, le=100)] = 10,
    tags: Annotated[list[str], Query()] = [],   # 多值参数：?tags=a&tags=b
):
    return {"q": q, "page": page, "size": size, "tags": tags}

字符串校验：min_length、max_length、pattern（正则）。
list[str] 类型的查询参数必须显式写 Query()，否则会被当成请求体。
参数别名：Query(alias="item-query")，适用于 URL 中的参数名不是合法 Python 变量名的情况。

请求体数据

from typing import Optional, List
from pydantic import BaseModel, Field, field_validator
import uvicorn
from fastapi import FastAPI

# 嵌套模型
class Address(BaseModel):
    city: str

# 当类型不一致的时候，默认会进行自动转换，转换不一致的时候，会报错
class User(BaseModel):
    name: str
    age: int = Field(default=18, gt=0, lt=20, description="用户年龄必须大于0小于20之间")
    friends: Optional[List[int]] = None
    # 注意：Optional 只表示"可以传 null"，没有默认值的字段仍然是必填的
    description: Optional[str] = Field(default=None, description="用户描述")
    address: Address

    # 自定义字段验证器
    @field_validator("name")
    @classmethod
    def check_name(cls, value: str) -> str:
        if len(value) < 3:
            raise ValueError("用户名必须不少于3个字符")
        return value

app = FastAPI()

@app.post("/user/register")
async def register(user: User):
    print("user.name:", user.name)
    print("user.age:", user.age)
    print("user.friends:", user.friends)
    return user

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

对应的请求体示例：

{
  "name": "zhangsan",
  "age": 18,
  "friends": [1, 2],
  "address": {"city": "北京"}
}

函数参数的类型是 BaseModel 子类时，FastAPI 会把它当作 JSON 请求体解析。
同一个函数里可以同时声明路径参数、查询参数和请求体，FastAPI 会自动区分：在路径中出现的是路径参数，单一类型（int、str 等）是查询参数，Pydantic 模型是请求体。
校验失败时自动返回 422，不需要手写校验逻辑。
在代码中把模型转回字典用 user.model_dump()，Pydantic v1 中对应的是 user.dict()。

form表单数据

接收表单数据用 Form，它依赖 python-multipart：

1	uv add python-multipart

import uvicorn
from fastapi import FastAPI, Form

app = FastAPI()


@app.post("/user/register")
async def register(username: str = Form()
                   , password: str = Form()):
    print("username:", username)
    print("password:", password)
    return {"message": "注册成功"}


if __name__ == "__main__":
    uvicorn.run("main:app", port=8000, reload=True)

表单的请求头是 application/x-www-form-urlencoded（或含文件时的 multipart/form-data），和 JSON 请求体不能在同一个接口中混用。

文件上传

文件上传同样依赖 python-multipart。FastAPI 提供了两种方式接收文件：

import uvicorn
from typing import Annotated
from fastapi import FastAPI, File, Form, UploadFile

app = FastAPI()


@app.post("/file")
async def upload_bytes(file: Annotated[bytes, File()]):
    # 整个文件内容以 bytes 读入内存，只适合小文件
    return {"file_size": len(file)}


@app.post("/upload")
async def upload_file(file: UploadFile, desc: Annotated[str, Form()] = ""):
    # UploadFile 使用"假脱机文件"，超过一定大小会写到磁盘，适合大文件
    content = await file.read()
    with open(f"statics/{file.filename}", "wb") as f:
        f.write(content)
    return {"filename": file.filename, "content_type": file.content_type, "desc": desc}


@app.post("/uploads")
async def upload_files(files: list[UploadFile]):
    # 多文件上传
    return {"filenames": [f.filename for f in files]}


if __name__ == "__main__":
    uvicorn.run("main:app", port=8000, reload=True)

UploadFile 常用属性和方法：

成员	说明
`filename`	上传时的原始文件名
`content_type`	MIME 类型，如 `image/png`
`size`	文件大小（字节）
`await read()` / `await write()`	读取 / 写入内容
`await seek(0)`	移动文件指针
`await close()`	关闭文件

生产环境中不要直接用客户端传来的 filename 作为保存路径，应重新生成文件名（如 uuid4().hex），并校验文件类型和大小，防止路径穿越与恶意文件。

除了自动解析的参数，也可以直接拿到原始请求对象，或者单独声明请求头、Cookie：

from typing import Annotated
from fastapi import FastAPI, Request, Header, Cookie

app = FastAPI()


@app.get("/request")
async def get_request(request: Request):
    return {
        "url": str(request.url),
        "method": request.method,
        "client_ip": request.client.host,
        "query_params": dict(request.query_params),
        "user_agent": request.headers.get("user-agent"),
    }


@app.get("/headers")
async def get_headers(
    user_agent: Annotated[str | None, Header()] = None,   # 自动把 user_agent 转成 User-Agent
    session_id: Annotated[str | None, Cookie()] = None,
):
    return {"user_agent": user_agent, "session_id": session_id}

请求静态文件

项目结构

my-fastapi-project/
├── .venv/                 # 虚拟环境（library root）
├── apps/                  # 应用模块目录
├── statics/               # 静态文件目录
│   └── test.txt
├── main.py                # 项目入口文件

import uvicorn
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# 访问 http://127.0.0.1:8000/static/test.txt 即可拿到 statics/test.txt
app.mount("/static", StaticFiles(directory="statics"), name="static")

if __name__ == "__main__":
    uvicorn.run("main:app", port=8000, reload=True)

响应模型相关的参数

响应模型的声明方式

使用返回类型注解

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
    tags: list[str] = []

@app.post("/items/")
async def create_item(item: Item) -> Item:  # 返回类型注解
    return item

@app.get("/items/")
async def read_items() -> list[Item]:  # 返回模型列表
    return [
        Item(name="Portal Gun", price=42.0),
        Item(name="Plumbus", price=32.0),
    ]

使用 `response_model` 装饰器参数

from typing import Any
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
    tags: list[str] = []

@app.post("/items/", response_model=Item)  # 装饰器参数
async def create_item(item: Item) -> Any:  # 返回类型声明为 Any
    return item  # 实际返回的是 Item 实例

@app.get("/items/", response_model=list[Item])
async def read_items() -> Any:
    return [  # 返回字典，但会被转换为 Item 模型
        {"name": "Portal Gun", "price": 42.0},
        {"name": "Plumbus", "price": 32.0},
    ]

数据过滤

from pydantic import BaseModel
import uvicorn
from fastapi import FastAPI

app = FastAPI()

# 输入模型：包含密码
class UserIn(BaseModel):
    username: str
    password: str
    email: str
    full_name: str | None = None


# 输出模型：不包含密码
class UserOut(BaseModel):
    username: str
    email: str
    full_name: str | None = None


@app.post("/user/")
async def create_user(user: UserIn) -> UserOut:
    # 即使函数返回包含密码的 user 对象
    # 返回类型注解 -> UserOut 也会自动过滤掉 password 字段
    return user


if __name__ == "__main__":
    uvicorn.run("main:app", port=8000, reload=True)

联合类型

当接口在不同条件下返回不同类型的响应时，可以使用 Union 类型：

from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class CarItem(BaseModel):
    description: str
    type: str = "car"

class PlaneItem(BaseModel):
    description: str
    type: str = "plane"
    size: int

items = {
    "item1": {"description": "A fast car", "type": "car"},
    "item2": {"description": "A big plane", "type": "plane", "size": 5},
}

@app.get("/items/{item_id}", response_model=Union[CarItem, PlaneItem])
async def read_item(item_id: str):
    return items[item_id]

FastAPI 会根据返回数据的实际结构自动匹配对应的响应模型。

响应模型其他编码参数

1. `response_model_exclude`：排除指定字段

@app.post("/items/", response_model=Item, response_model_exclude={"tax", "description"})
async def create_item(item: Item):
    return item
# 响应中不会包含 tax 和 description 字段

2. `response_model_include`：只包含指定字段

@app.post("/items/", response_model=Item, response_model_include={"name", "price"})
async def create_item(item: Item):
    return item
# 响应中只包含 name 和 price 字段

3. `response_model_exclude_none`：排除值为 `None` 的字段

@app.post("/items/", response_model=Item, response_model_exclude_none=True)
async def create_item(item: Item):
    return item
# 响应中不会包含值为 None 的字段

4. `response_model_exclude_unset`：排除未设置的字段

@app.post("/items/", response_model=Item, response_model_exclude_unset=True)
async def create_item(item: Item):
    return item
# 响应中不会包含未在请求中提供的字段

5. `response_model_exclude_defaults`：排除等于默认值的字段

@app.post("/items/", response_model=Item, response_model_exclude_defaults=True)
async def create_item(item: Item):
    return item
# 响应中不会包含值与默认值相同的字段

状态码与其他响应类型

指定状态码

from fastapi import FastAPI, status

app = FastAPI()

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

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    return None   # 204 不能带响应体

status 模块中定义了所有常用状态码常量，比直接写数字 201 可读性更好，编辑器也能自动补全。

返回其他类型的响应

默认情况下，FastAPI 会把返回值通过 jsonable_encoder 转换后包装成 JSONResponse。需要其他响应时，直接返回对应的 Response 子类即可：

from fastapi import FastAPI
from fastapi.responses import (
    JSONResponse, HTMLResponse, PlainTextResponse,
    RedirectResponse, FileResponse, StreamingResponse,
)

app = FastAPI()


@app.get("/json")
async def json_resp():
    # 自定义状态码和响应头
    return JSONResponse(content={"msg": "ok"}, status_code=200, headers={"X-Token": "abc"})


@app.get("/html", response_class=HTMLResponse)
async def html_resp():
    return "<h1>Hello FastAPI</h1>"


@app.get("/text", response_class=PlainTextResponse)
async def text_resp():
    return "plain text"


@app.get("/redirect")
async def redirect():
    return RedirectResponse(url="/docs")


@app.get("/download")
async def download():
    # filename 参数会设置 Content-Disposition，浏览器会弹出下载
    return FileResponse("statics/test.txt", filename="test.txt")


@app.get("/stream")
async def stream():
    async def gen():
        for i in range(5):
            yield f"line {i}\n"
    return StreamingResponse(gen(), media_type="text/plain")

直接返回 Response 对象时，FastAPI 不会再经过 response_model 做数据过滤和校验。

异常处理

HTTPException

业务中需要返回错误时，用 raise 抛出 HTTPException，而不是 return 它：

from fastapi import FastAPI, HTTPException

app = FastAPI()
items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "item-missing"},
        )
    return {"item": items[item_id]}

客户端收到的响应为 404，响应体是 {"detail": "Item not found"}。detail 可以是字符串，也可以是字典或列表。

自定义异常与全局处理器

项目里通常会定义自己的业务异常，再注册一个全局处理器，统一返回格式：

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()


class BizException(Exception):
    def __init__(self, code: int, msg: str):
        self.code = code
        self.msg = msg


@app.exception_handler(BizException)
async def biz_exception_handler(request: Request, exc: BizException):
    return JSONResponse(status_code=200, content={"code": exc.code, "msg": exc.msg, "data": None})


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    # 覆盖默认的 422 返回格式
    return JSONResponse(
        status_code=422,
        content={"code": 422, "msg": "参数校验失败", "data": exc.errors()},
    )


@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id != 1:
        raise BizException(code=1001, msg="用户不存在")
    return {"code": 0, "msg": "ok", "data": {"id": 1, "name": "zhangsan"}}

在任意深度的函数（包括依赖、service 层）中抛出的异常都能被全局处理器捕获，这样业务代码里只管 raise，不必层层返回错误码。

依赖注入

在 FastAPI 里，「依赖」是一个可调用对象（函数或类）。FastAPI 会在调用路径函数之前先执行它，再把返回值注入到参数中。公共参数复用、数据库会话管理、登录鉴权和权限校验一般都用依赖来做。

函数依赖

from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, page: int = 1, size: int = 10):
    # 依赖函数本身也可以声明查询参数、请求头等，FastAPI 会照常解析
    return {"q": q, "skip": (page - 1) * size, "limit": size}


# 用类型别名复用依赖声明
CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons

类作为依赖

class Pagination:
    def __init__(self, page: int = 1, size: int = 10):
        self.skip = (page - 1) * size
        self.limit = size


@app.get("/articles/")
async def read_articles(p: Annotated[Pagination, Depends()]):
    # Depends() 不传参数时，使用类型注解中的类作为依赖
    return {"skip": p.skip, "limit": p.limit}

带 `yield` 的依赖

yield 之前的代码在请求处理前执行，之后的代码在响应返回后执行，可以用来管理数据库连接这类需要释放的资源：

async def get_db():
    db = DBSession()
    try:
        yield db          # 注入到路径函数中
    finally:
        await db.close()  # 无论是否出现异常都会关闭


@app.get("/orders/")
async def read_orders(db: Annotated[DBSession, Depends(get_db)]):
    return await db.query_orders()

依赖嵌套与鉴权示例

依赖可以继续依赖其他依赖，FastAPI 会自动解析整条依赖链，同一个请求内相同的依赖只会执行一次，结果会被缓存。

from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException, APIRouter

app = FastAPI()

fake_tokens = {"token-admin": {"name": "admin", "is_admin": True},
               "token-user": {"name": "tom", "is_admin": False}}


async def get_current_user(authorization: Annotated[str | None, Header()] = None):
    if not authorization or authorization not in fake_tokens:
        raise HTTPException(status_code=401, detail="未登录或 token 无效")
    return fake_tokens[authorization]


async def require_admin(user: Annotated[dict, Depends(get_current_user)]):
    if not user["is_admin"]:
        raise HTTPException(status_code=403, detail="权限不足")
    return user


@app.get("/me")
async def me(user: Annotated[dict, Depends(get_current_user)]):
    return user


# 只需要依赖的副作用（校验），不关心返回值时，可以放在 dependencies 中
admin_router = APIRouter(prefix="/admin", dependencies=[Depends(require_admin)])


@admin_router.get("/dashboard")
async def dashboard():
    return {"msg": "欢迎进入后台"}


app.include_router(admin_router)

依赖可以挂在三个层级上：

单个路径函数的参数中，或装饰器的 dependencies=[...]；
APIRouter(dependencies=[...])，作用于该路由下所有接口；
FastAPI(dependencies=[...])，作用于全局所有接口。

真实项目中的 token 一般使用 JWT（如 pyjwt 库）签发和校验，FastAPI 也提供了 OAuth2PasswordBearer 等工具类，配合依赖注入即可实现完整的登录鉴权。

中间件与 CORS

自定义中间件

中间件会在每个请求到达路径函数之前、以及每个响应返回给客户端之前执行：

import time
from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)   # 交给后续的中间件和路径函数处理
    response.headers["X-Process-Time"] = f"{time.perf_counter() - start:.4f}"
    return response

有多个中间件时，后添加的先执行（洋葱模型）：请求从外到内，响应从内到外。

跨域（CORS）

前后端分离项目中，前端页面（如 http://localhost:5173）请求后端接口（http://localhost:8000）属于跨域，需要后端放行：

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

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],  # 允许的源，开发时可用 ["*"]
    allow_credentials=True,                   # 是否允许携带 Cookie
    allow_methods=["*"],                      # 允许的请求方法
    allow_headers=["*"],                      # 允许的请求头
)

allow_credentials=True 时，allow_origins 不能使用 ["*"]，必须写明具体的源。

生命周期与后台任务

lifespan 事件

需要在应用启动时初始化资源（数据库连接池、加载模型、读取配置），关闭时释放资源，可以使用 lifespan：

from contextlib import asynccontextmanager
from fastapi import FastAPI

resources = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    # yield 之前：应用启动时执行一次
    resources["db"] = "数据库连接池"
    print("应用启动")
    yield
    # yield 之后：应用关闭时执行一次
    resources.clear()
    print("应用关闭")


app = FastAPI(lifespan=lifespan)

旧版本中的 @app.on_event("startup") / @app.on_event("shutdown") 已经被标记为废弃，新项目统一使用 lifespan。

后台任务

发送邮件、写日志这类不需要等待结果的操作，可以交给 BackgroundTasks 在响应返回之后执行：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", "a", encoding="utf-8") as f:
        f.write(message + "\n")


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"notification sent to {email}")
    return {"message": "通知已在后台发送"}

BackgroundTasks 和应用运行在同一个进程中，适合轻量任务。耗时长、需要重试或分布式执行的任务，应使用 Celery、ARQ 等任务队列。

ORM

Tortoise-ORM 是一个受 Django ORM 启发、基于 asyncio 的轻量级 ORM，接口全部是异步的，可以直接在 async def 路径函数里 await。

准备工作与环境搭建

# 基础库 + 对应数据库的异步驱动，按需选择一个即可
uv add "tortoise-orm[aiosqlite]"   # SQLite
uv add "tortoise-orm[asyncpg]"     # PostgreSQL
uv add "tortoise-orm[asyncmy]"     # MySQL

项目结构：

my-fastapi-project/
├── app/
│   ├── __init__.py
│   ├── main.py          # 应用入口
│   ├── config.py        # 数据库配置
│   ├── models.py        # Tortoise ORM 模型定义
│   ├── schemas.py       # Pydantic 模型 (用于请求/响应验证)
│   └── api.py           # 路由与接口
├── migrations/          # aerich 生成的迁移文件
└── pyproject.toml

数据库配置

# app/config.py
TORTOISE_ORM = {
    "connections": {
        # 格式：数据库类型://用户名:密码@主机:端口/数据库名
        "default": "mysql://root:123456@127.0.0.1:3306/fastapi_demo",
        # SQLite 写法："sqlite://db.sqlite3"
        # PostgreSQL 写法："postgres://postgres:123456@127.0.0.1:5432/fastapi_demo"
    },
    "apps": {
        "models": {  # <-- 应用名称，模型关系中的 "models.Clas" 就是"应用名.模型名"
            "models": ["app.models", "aerich.models"],  # <-- 使用 aerich 时必须包含 aerich.models
            "default_connection": "default",
        },
    },
    "use_tz": False,
    "timezone": "Asia/Shanghai",
}

真实项目中数据库密码不要硬编码在代码里，应从环境变量或 .env 文件中读取（例如使用 pydantic-settings）。

创建模型类

以选课系统为例，app/models.py：

from tortoise.models import Model
from tortoise import fields


class Student(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=32, description="The name of the student")
    age = fields.IntField(description="The age of the student")
    pwd = fields.CharField(max_length=32, description="The password of the student")
    sno = fields.IntField(description="学号")

    # 一对多关系
    clas = fields.ForeignKeyField("models.Clas", related_name="students")

    # 多对多关系
    courses = fields.ManyToManyField("models.Course", related_name="students")


class Course(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=32, description="The name of the course")

    teacher = fields.ForeignKeyField("models.Teacher", related_name="courses")


class Teacher(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=32, description="The name of the teacher")
    pwd = fields.CharField(max_length=32, description="The password of the teacher")
    tno = fields.IntField(description="教师号")


class Clas(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=32, description="The name of the class")

常用字段类型：

字段	对应数据库类型	说明
`IntField` / `BigIntField`	`INT` / `BIGINT`	`pk=True` 表示主键，默认自增
`CharField`	`VARCHAR`	必须指定 `max_length`
`TextField`	`TEXT`	长文本
`BooleanField`	`BOOL`	布尔值
`DecimalField`	`DECIMAL`	需要 `max_digits`、`decimal_places`
`DatetimeField`	`DATETIME`	`auto_now_add=True` 创建时写入，`auto_now=True` 每次保存时更新
`JSONField`	`JSON`	存储字典或列表
`ForeignKeyField`	外键	一对多，`on_delete` 默认 `CASCADE`
`ManyToManyField`	中间表	多对多，自动创建中间表
`OneToOneField`	唯一外键	一对一

通用参数：null=True（允许为空）、default=xxx（默认值）、unique=True（唯一）、index=True（建索引）、description="..."（字段注释）。

示例为了聚焦 ORM 用法，pwd 字段直接存储了明文。真实项目中密码必须先哈希（如 bcrypt、passlib）再入库，并且绝不能出现在接口返回值中。

迁移

数据库迁移用 aerich。在 uv 项目里，所有 aerich 命令都要通过 uv run 执行，这样才会用到项目虚拟环境里的依赖。

安装 Aerich

1	uv add --dev aerich[toml]

--dev 把 aerich 记为开发依赖，生产环境不需要它；[toml] 扩展让 aerich 能读写 pyproject.toml 中的配置。

检查配置

确认前面 app/config.py 中的 TORTOISE_ORM 已经包含了 aerich.models，并记下应用名称（这里是 "models"），后续命令会用到。

初始化 Aerich

初始化 Aerich 环境。在项目根目录执行下面的命令，-t 参数是 Tortoise ORM 配置变量的导入路径（app/config.py 中的 TORTOISE_ORM）：
1
uv run aerich init -t app.config.TORTOISE_ORM
这条命令会创建 migrations 文件夹，并把配置写入 pyproject.toml 的 [tool.aerich] 中。
初始化数据库。下面的命令会在数据库中创建 aerich 表，并生成初始迁移文件：
1
uv run aerich init-db

日常迁移流程

每次修改 models.py 中的模型后，执行两步：

生成迁移文件，--name 写这次改了什么，例如 add_user_age_field：
1
uv run aerich migrate --name "描述你的修改"
执行迁移文件，把改动同步到数据库：
1
uv run aerich upgrade

其他常用命令

1
2
3

uv run aerich history              # 查看迁移历史
uv run aerich downgrade            # 回滚到上一个版本
uv run aerich downgrade -v <版本号>  # 回滚到指定版本，版本号从 history 中查

注册到 FastAPI

模型和迁移准备好后，还需要在应用启动时初始化 Tortoise、关闭时释放连接。借助前面讲的 lifespan 即可完成，app/main.py：

from contextlib import asynccontextmanager

import uvicorn
from fastapi import FastAPI
from tortoise.contrib.fastapi import RegisterTortoise

from app.api import router
from app.config import TORTOISE_ORM


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with RegisterTortoise(
        app,
        config=TORTOISE_ORM,
        generate_schemas=False,        # 表结构交给 aerich 管理，不自动建表
        add_exception_handlers=True,   # DoesNotExist 等 ORM 异常自动转换成 404 等响应
    ):
        yield


app = FastAPI(lifespan=lifespan)
app.include_router(router, prefix="/api")

if __name__ == "__main__":
    uvicorn.run("app.main:app", port=8000, reload=True)

在项目根目录启动：

1	uv run uvicorn app.main:app --reload

旧版本 Tortoise 使用 register_tortoise(app, config=TORTOISE_ORM) 注册，新版本推荐上面的 RegisterTortoise + lifespan 写法。不想用 aerich 的小项目，可以把 generate_schemas 设为 True，启动时自动建表（只会创建不存在的表，不会修改已有表结构）。

接口开发

下面的 app/api.py 用选课系统演示 Tortoise-ORM 的常用操作。

注意代码中路由的声明顺序：/students/filter、/students/count 这类固定路径必须写在 /students/{student_id} 之前，否则 filter、count 会被当作 student_id 解析而返回 422。所以「单条查改删」放在了文件最后。

"""
Tortoise-ORM 教学演示路由

本文件用 4 个模型演示 Tortoise-ORM 的常用操作。

模型关系速览：

  Teacher(教师) ──1:N──> Course(课程) ──M:N──> Student(学生) ──N:1──> Clas(班级)

  具体字段与 related_name：
    Student.clas   = FK("models.Clas",   related_name="students")
      → 正向: student.clas     拿到班级对象
      → 反向: clas.students    拿到该班所有学生（QuerySet）

    Student.courses = M2M("models.Course", related_name="students")
      → 正向: student.courses  拿到该生所选课程（QuerySet）
      → 反向: course.students  拿到选了该课的所有学生（QuerySet）

    Course.teacher  = FK("models.Teacher", related_name="courses")
      → 正向: course.teacher   拿到教师对象
      → 反向: teacher.courses  拿到该教师所教课程（QuerySet）

  注意：related_name 就是"反向关联的名字"，定义在"多"的那一侧，
    但挂在"一"的那一侧使用，初学时很容易搞混。
"""
from fastapi import APIRouter, HTTPException, Body
from tortoise.expressions import Q
from tortoise.functions import Count
from pydantic import BaseModel
from typing import Optional
from app.models import Student, Course, Teacher, Clas

router = APIRouter()


# ===== 1. 基础 CRUD：新增（单条查改删见第 10 节） =====

class StudentIn(BaseModel):
    name: str
    age: int
    pwd: str
    sno: int
    clas_id: int


class CourseIn(BaseModel):
    name: str
    teacher_id: int


class TeacherIn(BaseModel):
    name: str
    pwd: str
    tno: int


class ClasIn(BaseModel):
    name: str


class StudentUpdate(BaseModel):
    name: Optional[str] = None
    age: Optional[int] = None


@router.post("/students", tags=["1-基础CRUD"])
async def create_student(data: StudentIn):
    """
    【接口用途】：新增一名学生
    【演示特性】：Model.create()，直接创建并插入数据库，返回已含 id 的新对象
    【底层SQL】：INSERT INTO student (name, age, pwd, sno, clas_id) VALUES (?, ?, ?, ?, ?)
    【初学提示】：create() 是"构造 + 保存"一步到位，等价于 Student(**kwargs) 再 .save()
    """
    student = await Student.create(
        name=data.name, age=data.age, pwd=data.pwd,
        sno=data.sno, clas_id=data.clas_id,
    )
    return {"id": student.id, "name": student.name}


# ===== 2. 批量操作 =====

@router.post("/students/bulk", tags=["2-批量操作"])
async def bulk_create_students(data: list[StudentIn] = Body(...)):
    """
    【接口用途】：一次性批量导入多名学生（如新生入学批量录入）
    【演示特性】：Model.bulk_create()，一次调用插入多条记录
    【底层SQL】：多条 INSERT INTO student (...) VALUES (...)（SQLite 不支持多行 INSERT）
    【初学提示】：bulk_create 比循环 create 快，因为减少了 Python↔数据库的往返次数。
      注意：传入的是"未保存的 Model 实例列表"，不是字典列表。
    """
    objects = [
        Student(name=d.name, age=d.age, pwd=d.pwd, sno=d.sno, clas_id=d.clas_id)
        for d in data
    ]
    await Student.bulk_create(objects)
    return {"created_count": len(objects)}


@router.put("/students/bulk-age", tags=["2-批量操作"])
async def bulk_update_student_age(ids: list[int] = Body(...), age: int = Body(...)):
    """
    【接口用途】：批量修改多名学生的年龄（如统一升级）
    【演示特性】：Model.bulk_update()，批量修改指定字段
    【底层SQL】：逐条 UPDATE student SET age=? WHERE id=?
    【初学提示】：步骤：1) 查出对象列表 → 2) 在 Python 中修改属性 → 3) bulk_update(list, ["字段名"])
      第二个参数 ["age"] 告诉 ORM 只更新 age 字段，其他字段不动。
    """
    students = await Student.filter(id__in=ids)
    for s in students:
        s.age = age
    await Student.bulk_update(students, ["age"])
    return {"updated_count": len(students)}


@router.post("/courses/bulk", tags=["2-批量操作"])
async def bulk_create_courses(data: list[CourseIn] = Body(...)):
    """
    【接口用途】：批量创建课程（如新学期排课）
    【演示特性】：bulk_create() 同样适用于 Course 等任意模型
    【底层SQL】：多条 INSERT INTO course (name, teacher_id) VALUES (?, ?)
    """
    objects = [Course(name=d.name, teacher_id=d.teacher_id) for d in data]
    await Course.bulk_create(objects)
    return {"created_count": len(objects)}


# ===== 3. 高级过滤与查询 =====

@router.get("/students/filter", tags=["3-高级过滤"])
async def filter_students(
    min_age: Optional[int] = None,
    max_age: Optional[int] = None,
    clas_id: Optional[int] = None,
    order: str = "id",
    limit: int = 20,
    offset: int = 0,
):
    """
    【接口用途】：按年龄范围、班级等条件筛选学生，支持排序和分页
    【演示特性】：链式调用 .filter().order_by().limit().offset()
    【底层SQL】：SELECT ... FROM student WHERE age>=? AND age<=? AND clas_id=? ORDER BY id LIMIT 20 OFFSET 0
    【初学提示】：Tortoise 的链式调用是"惰性求值"，调用 .filter() 只是构建查询条件，
      并不会真的去查数据库，直到 await 时才拼接成一条 SQL 发出。
    """
    qs = Student.all()
    if min_age is not None:
        qs = qs.filter(age__gte=min_age)  # __gte → WHERE age >= ?
    if max_age is not None:
        qs = qs.filter(age__lte=max_age)  # __lte → WHERE age <= ?
    if clas_id is not None:
        qs = qs.filter(clas_id=clas_id)
    students = await qs.order_by(order).limit(limit).offset(offset)
    return {"count": len(students), "students": [s.name for s in students]}


@router.get("/teachers/search", tags=["3-高级过滤"])
async def search_teachers(keyword: Optional[str] = None):
    """
    【接口用途】：按姓名模糊搜索教师（如输入"张"搜出所有张老师）
    【演示特性】：__icontains，不区分大小写的模糊匹配
    【底层SQL】：SELECT ... FROM teacher WHERE name LIKE '%张%'（SQLite 对 ASCII 不敏感）
    【初学提示】：__icontains 中的 i 代表 insensitive（不敏感），contains 代表包含。
      常用查找运算符：__exact(精确)、__contains(区分大小写包含)、__icontains(不区分)。
    """
    if not keyword:
        teachers = await Teacher.all()
    else:
        teachers = await Teacher.filter(name__icontains=keyword)
    return {"teachers": [t.name for t in teachers]}


@router.get("/courses/search", tags=["3-高级过滤"])
async def search_courses(keyword: Optional[str] = None, min_teacher_tno: Optional[int] = None):
    """
    【接口用途】：按课程名模糊搜索，并可限定授课教师的教师号下限
    【演示特性】：__icontains + 跨表 __gte 组合过滤
    【底层SQL】：SELECT ... FROM course WHERE name LIKE '%数学%' AND teacher.tno >= 100
    【初学提示】：teacher__tno__gte 意思是"跨到 teacher 表，取 tno 字段，大于等于"。
      双下划线 __ 是 Tortoise 跨表的语法分隔符，每一段代表一层关系跳转。
    """
    qs = Course.all()
    if keyword:
        qs = qs.filter(name__icontains=keyword)
    if min_teacher_tno is not None:
        qs = qs.filter(teacher__tno__gte=min_teacher_tno)
    courses = await qs
    return {"courses": [{"id": c.id, "name": c.name} for c in courses]}


@router.get("/students/by-classes", tags=["3-高级过滤"])
async def students_by_classes(clas_ids: str):
    """
    【接口用途】：查询属于多个班级的所有学生（如查"1班和2班"的学生）
    【演示特性】：__in，SQL 的 IN (v1, v2, ...) 多值筛选
    【底层SQL】：SELECT ... FROM student WHERE clas_id IN (1, 2)
    【初学提示】：clas_ids 传逗号分隔字符串如 "1,2,3"，代码中 split 后转成列表。
      __in 只接受列表/元组/集合，不能传字符串。
    """
    ids = [int(x) for x in clas_ids.split(",")]
    students = await Student.filter(clas_id__in=ids)
    return {"students": [s.name for s in students]}


@router.get("/students/exclude", tags=["3-高级过滤"])
async def exclude_students(min_age: int = 0):
    """
    【接口用途】：查询年龄不在指定范围内的学生（如排除 18 岁以下的）
    【演示特性】：.exclude()，排除符合条件的记录，生成 NOT 子句
    【底层SQL】：SELECT ... FROM student WHERE NOT (age >= ?)
    【初学提示】：.exclude() 和 .filter() 用法一样，只是逻辑取反。
      多个 .exclude() 之间是 AND 关系，不是 OR。
    """
    students = await Student.all().exclude(age__gte=min_age)
    return {"students": [s.name for s in students]}


# ===== 4. Q 对象：复杂逻辑组合（OR / AND / NOT） =====

@router.get("/students/q-search", tags=["4-Q对象"])
async def q_search_students(keyword: Optional[str] = None, min_age: Optional[int] = None):
    """
    【接口用途】：按姓名模糊搜索 OR 按学号精确搜索，并可限定最低年龄
    【演示特性】：Q() 对象实现 OR 条件，一个 .filter() 里同时满足"或"逻辑
    【底层SQL】：WHERE (name LIKE '%张%' OR sno = 2021001) AND age >= 18
    【初学提示】：普通 .filter(a=1, b=2) 是 AND 关系，无法表达 OR。
      Q(a=1) | Q(b=2) 就是 OR；Q(a=1) & Q(b=2) 就是 AND；~Q(a=1) 就是 NOT。
      Q 对象可以和普通 .filter() 混用，后者自动 AND 上去。
    """
    qs = Student.all()
    if keyword:
        cond = Q(name__icontains=keyword)
        if keyword.isdigit():  # sno 是整数字段，只有纯数字关键词才参与学号匹配
            cond |= Q(sno=int(keyword))
        qs = qs.filter(cond)
    if min_age is not None:
        qs = qs.filter(age__gte=min_age)
    students = await qs
    return {"students": [s.name for s in students]}


@router.get("/teachers/q-filter", tags=["4-Q对象"])
async def q_filter_teachers(name_kw: Optional[str] = None, tno_min: Optional[int] = None):
    """
    【接口用途】：查询姓名含关键词 AND 教师号不低于某值的教师
    【演示特性】：Q() & Q()，AND 组合（演示语法，实际此处等价于链式 .filter()）
    【底层SQL】：WHERE name LIKE '%李%' AND tno >= 100
    【初学提示】：Q() & Q() 的 AND 组合在语义上和两次 .filter() 等价，
      但 Q 对象的优势在于可以和 | (OR) 混用，如 (Q(a=1) & Q(b=2)) | Q(c=3)。
    """
    qs = Teacher.all()
    if name_kw and tno_min is not None:
        qs = qs.filter(Q(name__icontains=name_kw) & Q(tno__gte=tno_min))
    elif name_kw:
        qs = qs.filter(name__icontains=name_kw)
    teachers = await qs
    return {"teachers": [t.name for t in teachers]}


# ===== 5. related_name 正向/反向查询 =====

@router.get("/related-name-explained", tags=["5-related_name详解"])
async def related_name_explained():
    """
    【接口用途】：一次性演示所有 related_name 的正向和反向用法
    【演示特性】：related_name 决定了"从对方模型反查"时用的属性名

    related_name 核心规则：

    定义在模型 A 的 ForeignKey/M2M 字段上：
      field = ForeignKeyField("models.B", related_name="xxx")

    含义：
      正向：A 的实例 → a.field     → 拿到 B 的对象（FK）或 QuerySet（M2M）
      反向：B 的实例 → b.xxx       → 拿到关联 A 的 QuerySet

    本项目的三个 related_name：

    ① Student.clas = FK("models.Clas", related_name="students")
       正向：student.clas   → 班级对象（一个 Clas 实例）
       反向：clas.students  → 该班所有学生（QuerySet[Student]）

    ② Student.courses = M2M("models.Course", related_name="students")
       正向：student.courses → 该生所选课程（QuerySet[Course]）
       反向：course.students → 选了该课的学生（QuerySet[Student]）

    ③ Course.teacher = FK("models.Teacher", related_name="courses")
       正向：course.teacher  → 教师对象（一个 Teacher 实例）
       反向：teacher.courses → 该教师所教课程（QuerySet[Course]）

    初学避坑：
      - related_name 写在"多"方（Student），但挂在"一"方（Clas）上用
      - 反向查出来的是 QuerySet，需要 await 才能拿到数据
      - 如果忘了设 related_name，Tortoise 会自动生成默认名，但不推荐依赖默认值
    """
    example = {
        "Student→Clas 正向": "student.clas → Clas 对象",
        "Clas→Student 反向": "clas.students → QuerySet[Student]",
        "Student→Course 正向": "student.courses → QuerySet[Course]",
        "Course→Student 反向": "course.students → QuerySet[Student]",
        "Course→Teacher 正向": "course.teacher → Teacher 对象",
        "Teacher→Course 反向": "teacher.courses → QuerySet[Course]",
    }
    return {"related_name规则": example}


@router.get("/students/{student_id}/forward-fk", tags=["5-related_name详解"])
async def student_forward_fk(student_id: int):
    """
    【接口用途】：查一名学生，并展示正向 FK 查询，通过 student.clas 拿到班级
    【演示特性】：正向 FK，用字段名直接访问关联对象
    【底层SQL】：先 SELECT ... FROM student WHERE id=?，再 SELECT ... FROM clas WHERE id=?
    【初学提示】：student.clas 是一个 Clas 对象，可以直接 .name 取属性。
      但注意：如果没做预加载，访问 student.clas 会额外发一条 SQL（N+1 问题）。
      预加载方案见下一节 select_related / prefetch_related。
    """
    student = await Student.get_or_none(id=student_id)
    if not student:
        raise HTTPException(404, "学生不存在")
    clas = await student.clas  # 正向 FK：字段名 clas → 拿到班级对象
    return {"student": student.name, "class": clas.name if clas else None}


@router.get("/classes/{clas_id}/reverse-fk", tags=["5-related_name详解"])
async def clas_reverse_fk(clas_id: int):
    """
    【接口用途】：查一个班级，并展示反向 FK 查询，通过 clas.students 拿到所有学生
    【演示特性】：反向 FK，用 related_name 访问关联的多个对象
    【底层SQL】：先 SELECT ... FROM clas WHERE id=?，再 SELECT ... FROM student WHERE clas_id=?
    【初学提示】：clas.students 是 QuerySet，需要 await 才能拿到列表。
      这就是 related_name="students" 的作用，让 Clas 模型凭空多出 .students 属性。
    """
    clas = await Clas.get_or_none(id=clas_id)
    if not clas:
        raise HTTPException(404, "班级不存在")
    students = await clas.students.all()  # 反向 FK：related_name="students" → QuerySet
    return {"class": clas.name, "students": [s.name for s in students]}


@router.get("/students/{student_id}/forward-m2m", tags=["5-related_name详解"])
async def student_forward_m2m(student_id: int):
    """
    【接口用途】：查一名学生，展示正向 M2M，通过 student.courses 拿到所选课程
    【演示特性】：正向 M2M，字段名 courses 直接就是 QuerySet
    【底层SQL】：SELECT ... FROM course JOIN student_course ON ... WHERE student_course.student_id=?
    【初学提示】：M2M 的正向和反向都是 QuerySet，因为多对多双方都是"多"。
      student.courses.add(course) 选课，student.courses.remove(course) 退课。
    """
    student = await Student.get_or_none(id=student_id)
    if not student:
        raise HTTPException(404, "学生不存在")
    courses = await student.courses.all()  # 正向 M2M：字段名 courses → QuerySet
    return {"student": student.name, "courses": [c.name for c in courses]}


@router.get("/courses/{course_id}/reverse-m2m", tags=["5-related_name详解"])
async def course_reverse_m2m(course_id: int):
    """
    【接口用途】：查一门课程，展示反向 M2M，通过 course.students 拿到选课学生
    【演示特性】：反向 M2M，related_name="students" 让 Course 拥有 .students 属性
    【底层SQL】：SELECT ... FROM student JOIN student_course ON ... WHERE student_course.course_id=?
    【初学提示】：虽然 Course 模型里没有写 students 字段，但 Student.courses 的
      related_name="students" 让 Tortoise 自动给 Course 加上了 .students 反向关联。
    """
    course = await Course.get_or_none(id=course_id)
    if not course:
        raise HTTPException(404, "课程不存在")
    students = await course.students.all()  # 反向 M2M：related_name="students"
    return {"course": course.name, "students": [s.name for s in students]}


# ===== 6. 关联查询与性能优化：select_related / prefetch_related =====

@router.get("/courses/with-teacher", tags=["6-关联预加载"])
async def courses_with_teacher():
    """
    【接口用途】：查询所有课程，同时带出每门课的授课教师姓名
    【演示特性】：select_related("teacher")，正向 FK 预加载，一次 JOIN 查完
    【底层SQL】：SELECT ... FROM course LEFT OUTER JOIN teacher ON course.teacher_id=teacher.id
    【初学提示】：不用 select_related 时，查 N 门课 → 再查 N 次教师 → 共 N+1 条 SQL。
      加了 select_related 后，只需 1 条 JOIN SQL，性能大幅提升。
      注意：select_related 只能用于 ForeignKey / OneToOne 的正向关系（"多对一"方向）。
    """
    courses = await Course.all().select_related("teacher")
    return [
        {"course": c.name, "teacher": c.teacher.name if c.teacher else None}
        for c in courses
    ]


@router.get("/students/with-class", tags=["6-关联预加载"])
async def students_with_class():
    """
    【接口用途】：查询所有学生，同时带出每个学生所属的班级名
    【演示特性】：select_related("clas")，学生→班级 正向 FK 预加载
    【底层SQL】：SELECT ... FROM student LEFT OUTER JOIN clas ON student.clas_id=clas.id
    【初学提示】：select_related 的参数是 FK 字段名（clas），不是 related_name。
      预加载后，student.clas 已经在内存中，访问时不再额外发 SQL。
    """
    students = await Student.all().select_related("clas")
    return [
        {"student": s.name, "class": s.clas.name if s.clas else None}
        for s in students
    ]


@router.get("/classes/with-students", tags=["6-关联预加载"])
async def classes_with_students():
    """
    【接口用途】：查询所有班级，同时带出每个班级的所有学生
    【演示特性】：prefetch_related("students")，反向一对多预加载
    【底层SQL】：① SELECT ... FROM clas  ② SELECT ... FROM student WHERE clas_id IN (1,2,...)
    【初学提示】：prefetch_related 发 2 条 SQL：先查主表，再一次性查所有关联数据。
      比 N+1 的 N 次查询好得多。参数 "students" 是 related_name，不是 FK 字段名。
      注意：prefetch_related 用于反向关系（一对多、多对多），select_related 用于正向 FK。
    """
    classes = await Clas.all().prefetch_related("students")
    return [
        {"class": c.name, "students": [s.name for s in c.students]}
        for c in classes
    ]


@router.get("/students/with-courses", tags=["6-关联预加载"])
async def students_with_courses():
    """
    【接口用途】：查询所有学生，同时带出每个学生已选的课程列表
    【演示特性】：prefetch_related("courses")，M2M 正向预加载
    【底层SQL】：① SELECT ... FROM student  ② 查中间表  ③ SELECT ... FROM course WHERE id IN (...)
    【初学提示】：M2M 的 prefetch_related 会发 3 条 SQL（主表 + 中间表 + 关联表），
      但仍比对每个学生单独 .courses.all() 的 N+1 模式高效得多。
    """
    students = await Student.all().prefetch_related("courses")
    return [
        {"student": s.name, "courses": [c.name for c in s.courses]}
        for s in students
    ]


@router.get("/teachers/with-courses", tags=["6-关联预加载"])
async def teachers_with_courses():
    """
    【接口用途】：查询所有教师，同时带出每位教师所教的所有课程
    【演示特性】：prefetch_related("courses")，反向一对多预加载
    【底层SQL】：① SELECT ... FROM teacher  ② SELECT ... FROM course WHERE teacher_id IN (1,2,...)
    【初学提示】：Teacher 模型里并没有写 courses 字段！
      但 Course.teacher 的 related_name="courses" 让 Tortoise 自动在 Teacher 上
      生成了 .courses 反向关联，所以 prefetch_related("courses") 可以直接用。
    """
    teachers = await Teacher.all().prefetch_related("courses")
    return [
        {"teacher": t.name, "courses": [c.name for c in t.courses]}
        for t in teachers
    ]


# ===== 7. 跨表关联过滤：用双下划线 __ 穿越关系链 =====

@router.get("/students/by-teacher", tags=["7-跨表过滤"])
async def students_by_teacher(teacher_name: str):
    """
    【接口用途】：查询选了某位老师所教课程的所有学生（如"查选了张老师课的学生"）
    【演示特性】：courses__teacher__name，双下划线穿越 M2M + FK 两层关系

    关系链拆解：
    Student →[courses]→ Course →[teacher]→ Teacher → .name

    第1步：courses__    从 Student 出发，沿 M2M 字段 courses 跳到 Course 表
    第2步：teacher__    从 Course 出发，沿 FK 字段 teacher 跳到 Teacher 表
    第3步：name=        在 Teacher 表上筛选 name = "张老师"

    【底层SQL】：SELECT student.* FROM student
      JOIN student_course ON student.id = student_course.student_id
      JOIN course ON student_course.course_id = course.id
      JOIN teacher ON course.teacher_id = teacher.id
      WHERE teacher.name = '张老师'
    【初学提示】：.distinct() 用来去重：同一老师教多门课，同一学生可能选了其中多门，
      不加 distinct 会出现重复。跨表过滤产生 INNER JOIN，未选课的学生不会出现。
    """
    students = await Student.filter(courses__teacher__name=teacher_name).distinct()
    return {"teacher": teacher_name, "students": [s.name for s in students]}


@router.get("/courses/by-class", tags=["7-跨表过滤"])
async def courses_by_class(clas_name: str):
    """
    【接口用途】：查询某个班级的学生选了哪些课程（如"查1班学生选的所有课"）
    【演示特性】：students__clas__name，从 Course 反向穿越 M2M + FK

    关系链拆解：
    Course →[students]→ Student →[clas]→ Clas → .name

    第1步：students__   从 Course 出发，沿 related_name="students" 跳到 Student 表
    第2步：clas__       从 Student 出发，沿 FK 字段 clas 跳到 Clas 表
    第3步：name=        在 Clas 表上筛选 name = "1班"

    【底层SQL】：SELECT course.* FROM course
      JOIN student_course ON course.id = student_course.course_id
      JOIN student ON student_course.student_id = student.id
      JOIN clas ON student.clas_id = clas.id
      WHERE clas.name = '1班'
    【初学提示】：students__ 用的是 related_name，不是字段名。
      跨表时正向用字段名，反向用 related_name，和预加载的参数规则一致。
    """
    courses = await Course.filter(students__clas__name=clas_name).distinct()
    return {"class": clas_name, "courses": [c.name for c in courses]}


@router.get("/students/by-course-name", tags=["7-跨表过滤"])
async def students_by_course_name(course_name: str):
    """
    【接口用途】：查询选了某门课（模糊匹配课程名）的所有学生
    【演示特性】：courses__name__icontains，跨 M2M + 模糊匹配

    关系链拆解：
    Student →[courses]→ Course → .name (模糊匹配)

    第1步：courses__       从 Student 出发，沿 M2M 字段 courses 跳到 Course 表
    第2步：name__icontains 在 Course 表上做模糊匹配

    【底层SQL】：SELECT student.* FROM student
      JOIN student_course ON student.id = student_course.student_id
      JOIN course ON student_course.course_id = course.id
      WHERE course.name LIKE '%数学%'
    【初学提示】：跨表过滤和跨表预加载的 __ 语法完全一致，
      区别：预加载是"把关联数据一起查回来"，过滤是"用关联数据做 WHERE 条件"。
    """
    students = await Student.filter(courses__name__icontains=course_name).distinct()
    return {"course_keyword": course_name, "students": [s.name for s in students]}


@router.get("/courses/by-teacher-tno", tags=["7-跨表过滤"])
async def courses_by_teacher_tno(min_tno: int):
    """
    【接口用途】：查询教师号不低于某值的所有教师所教的课程
    【演示特性】：teacher__tno__gte，跨一层 FK + 范围查找

    关系链拆解：
    Course →[teacher]→ Teacher → .tno (>= min_tno)

    第1步：teacher__   从 Course 出发，沿 FK 字段 teacher 跳到 Teacher 表
    第2步：tno__gte    在 Teacher 表上筛选 tno >= min_tno

    【底层SQL】：SELECT course.* FROM course
      JOIN teacher ON course.teacher_id = teacher.id
      WHERE teacher.tno >= 100
    【初学提示】：这是最简单的跨表过滤，只跨一层 FK。
      理解了这一层，多层跨表就是不断用 __ 往下接而已。
    """
    courses = await Course.filter(teacher__tno__gte=min_tno)
    return {"courses": [{"id": c.id, "name": c.name} for c in courses]}


# ===== 8. 聚合与分组统计 =====

@router.get("/classes/student-count", tags=["8-聚合统计"])
async def class_student_count():
    """
    【接口用途】：统计每个班级各有多少学生
    【演示特性】：.annotate(student_count=Count("students"))，分组计数
    【底层SQL】：SELECT clas.*, COUNT(student.id) AS student_count
      FROM clas LEFT OUTER JOIN student ON student.clas_id = clas.id GROUP BY clas.id
    【初学提示】：annotate 的参数名 "students" 是 related_name，和 prefetch_related 一致。
      annotate 不会单独发 SQL，而是把 COUNT 嵌入主查询的 SELECT 和 GROUP BY 中。
      结果中每个班级对象会多出一个 .student_count 属性。
    """
    classes = await Clas.all().annotate(student_count=Count("students"))
    return [
        {"class": c.name, "student_count": c.student_count}
        for c in classes
    ]


@router.get("/teachers/course-count", tags=["8-聚合统计"])
async def teacher_course_count():
    """
    【接口用途】：统计每位教师各教多少门课
    【演示特性】：.annotate(course_count=Count("courses"))，related_name 反向计数
    【底层SQL】：SELECT teacher.*, COUNT(course.id) AS course_count
      FROM teacher LEFT OUTER JOIN course ON course.teacher_id = teacher.id GROUP BY teacher.id
    【初学提示】：未带课的教师 course_count 为 0，因为 LEFT JOIN 保留了左表所有行。
      如果用 INNER JOIN 则不会出现 0，但 Tortoise 的 annotate 默认用 LEFT JOIN。
    """
    teachers = await Teacher.all().annotate(course_count=Count("courses"))
    return [
        {"teacher": t.name, "course_count": t.course_count}
        for t in teachers
    ]


@router.get("/students/values", tags=["8-聚合统计"])
async def student_values():
    """
    【接口用途】：只取所有学生的姓名和年龄，不要其他字段，也不要 ORM 对象
    【演示特性】：.values("name", "age")，返回字典列表，不构造 Model 实例
    【底层SQL】：SELECT student.name, student.age FROM student
    【初学提示】：values() 比 all() 轻量，省去构造 ORM 对象的开销，
      返回 [{"name": "张三", "age": 20}, ...] 纯字典，适合只读场景。
    """
    rows = await Student.all().values("name", "age")
    return {"students": rows}


@router.get("/students/values-list", tags=["8-聚合统计"])
async def student_values_list():
    """
    【接口用途】：只取所有学生的姓名和年龄，返回元组列表（比 values 更轻量）
    【演示特性】：.values_list("name", "age")，返回元组列表
    【底层SQL】：SELECT student.name, student.age FROM student
    【初学提示】：返回 [("张三", 20), ("李四", 21), ...]，字段顺序与参数一致。
      如果只取一个字段，可以加 flat=True 直接返回值列表：["张三", "李四", ...]。
    """
    rows = await Student.all().values_list("name", "age")
    return {"students": rows}


@router.get("/classes/student-count-values", tags=["8-聚合统计"])
async def class_student_count_values():
    """
    【接口用途】：统计各班人数，只返回班级名和人数的字典，不要 ORM 对象
    【演示特性】：annotate + values，聚合结果也用轻量字典返回
    【底层SQL】：SELECT clas.name, COUNT(student.id) AS student_count
      FROM clas LEFT JOIN student ... GROUP BY clas.id
    【初学提示】：annotate 算出 student_count，values 只取 name 和 student_count，
      结果是 [{"name": "1班", "student_count": 30}, ...]，适合直接返回给前端。
    """
    rows = await Clas.all().annotate(student_count=Count("students")).values("name", "student_count")
    return {"classes": rows}


# ===== 9. 轻量判断与去重 =====

@router.get("/students/{student_id}/enrolled/{course_id}", tags=["9-轻量判断"])
async def check_enrollment(student_id: int, course_id: int):
    """
    【接口用途】：判断某学生是否选了某门课（选课前查重）
    【演示特性】：.exists()，只返回 True/False，不加载任何数据
    【底层SQL】：SELECT 1 FROM student_course WHERE student_id=? AND course_id=? LIMIT 1
    【初学提示】：exists() 是最轻量的存在性检查，不构造 ORM 对象，不返回数据。
      对比"查出所有再判断 len > 0"的做法，exists() 只发一条 LIMIT 1 的 SQL。
    """
    student = await Student.get_or_none(id=student_id)
    if not student:
        raise HTTPException(404, "学生不存在")
    enrolled = await student.courses.filter(id=course_id).exists()
    return {"enrolled": enrolled}


@router.get("/students/count", tags=["9-轻量判断"])
async def student_count(clas_id: Optional[int] = None):
    """
    【接口用途】：统计学生总数，可按班级筛选
    【演示特性】：.count()，只返回数字，不加载任何对象
    【底层SQL】：SELECT COUNT(*) FROM student [WHERE clas_id=?]
    【初学提示】：count() 比 all() 再 len() 高效得多，后者要把全部数据加载到内存。
      如果只需要数字，永远用 count()。
    """
    qs = Student.all()
    if clas_id is not None:
        qs = qs.filter(clas_id=clas_id)
    total = await qs.count()
    return {"count": total}


@router.get("/courses/teacher-names", tags=["9-轻量判断"])
async def course_teacher_names():
    """
    【接口用途】：获取所有授课教师的姓名（去重）
    【演示特性】：.distinct() + .values()，去重 + 轻量字典
    【底层SQL】：SELECT DISTINCT teacher.name FROM course JOIN teacher ON ...
    【初学提示】：values("teacher__name") 用 __ 跨表取字段，会自动 JOIN，不需要 select_related。
      同一教师教多门课时，不加 distinct 会重复出现。
      distinct() 作用于整行结果，配合 values() 可对选定字段去重。
    """
    rows = await Course.all().distinct().values("teacher__name")
    return {"teacher_names": rows}


@router.get("/students/exists", tags=["9-轻量判断"])
async def student_exists(sno: int):
    """
    【接口用途】：判断某个学号是否已存在（注册前查重）
    【演示特性】：.filter().exists()，按业务字段判断存在性
    【底层SQL】：SELECT 1 FROM student WHERE sno=? LIMIT 1
    【初学提示】：和 get_or_none() 相比，exists() 不构造 Model 实例，只返回布尔值，
      更轻量。如果只需要知道"有没有"，不需要"拿到数据"，就用 exists()。
    """
    exists = await Student.filter(sno=sno).exists()
    return {"sno": sno, "exists": exists}


# ===== 10. 单条查改删：带 {student_id} 的路由放在最后，避免抢先匹配固定路径 =====

@router.get("/students/{student_id}", tags=["10-单条查改删"])
async def get_student(student_id: int):
    """
    【接口用途】：根据 id 查一名学生
    【演示特性】：Model.get_or_none()，找不到时返回 None，不抛异常
    【底层SQL】：SELECT ... FROM student WHERE id=? LIMIT 1
    【初学提示】：get() 找不到时会抛 DoesNotExist 异常。
      在接口开发中，get_or_none() + 手动判断更灵活，方便返回自定义 404 信息。
    """
    student = await Student.get_or_none(id=student_id)
    if not student:
        raise HTTPException(404, "学生不存在")
    return {"id": student.id, "name": student.name, "age": student.age}


@router.put("/students/{student_id}", tags=["10-单条查改删"])
async def update_student(student_id: int, data: StudentUpdate):
    """
    【接口用途】：修改学生的姓名或年龄（只改传入的字段）
    【演示特性】：model_dump(exclude_unset=True) + update_from_dict() + save()，部分字段更新
    【底层SQL】：UPDATE student SET name=?, age=? WHERE id=?
    【初学提示】：exclude_unset=True 只保留请求体中实际传了的字段，没传的不会被覆盖成 None。
      也可以用 await Student.filter(id=student_id).update(**fields)，
      区别是 filter().update() 不需要先查出对象，直接发 UPDATE 语句，更高效。
    """
    student = await Student.get_or_none(id=student_id)
    if not student:
        raise HTTPException(404, "学生不存在")
    update_fields = data.model_dump(exclude_unset=True)
    if update_fields:
        await student.update_from_dict(update_fields).save()
    return {"updated": update_fields}


@router.delete("/students/{student_id}", tags=["10-单条查改删"])
async def delete_student(student_id: int):
    """
    【接口用途】：删除一名学生
    【演示特性】：instance.delete()，删除数据库中的这条记录
    【底层SQL】：DELETE FROM student WHERE id=?
    【初学提示】：删除后 Python 中的 student 对象仍然存在（内存中），
      只是数据库里没了。也可以用 await Student.filter(id=xxx).delete() 批量删。
    """
    student = await Student.get_or_none(id=student_id)
    if not student:
        raise HTTPException(404, "学生不存在")
    await student.delete()
    return {"deleted": student_id}

接口测试

除了在 /docs 页面手动点击调试，FastAPI 还提供了 TestClient，可以配合 pytest 编写自动化测试，不需要真正启动服务。

1. 安装测试依赖

1	uv add --dev pytest httpx

2. 编写测试

假设 main.py 中有如下接口：

from fastapi import FastAPI, HTTPException

app = FastAPI()


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id != 1:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"id": 1, "name": "Portal Gun"}

新建 test_main.py：

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_read_item():
    response = client.get("/items/1")
    assert response.status_code == 200
    assert response.json() == {"id": 1, "name": "Portal Gun"}


def test_read_item_not_found():
    response = client.get("/items/2")
    assert response.status_code == 404
    assert response.json() == {"detail": "Item not found"}


def test_read_item_invalid_id():
    response = client.get("/items/abc")
    assert response.status_code == 422

3. 运行测试

1	uv run pytest -v

测试中需要替换掉某个依赖（例如把真实数据库换成测试库、跳过登录校验），可以使用 app.dependency_overrides[原依赖] = 替代依赖。

部署

开发时使用的 --reload 只适合本地调试，生产环境需要去掉它，并按 CPU 核数开启多个 worker 进程：

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

# 方式二：安装了 fastapi[standard] 后可用
uv run fastapi run app/main.py --port 8000 --workers 4

使用 Docker 部署时，可以基于 uv 官方镜像构建：

FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim

WORKDIR /app

# 先只复制依赖文件，利用 Docker 缓存加速构建
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev

COPY . .

EXPOSE 8000
CMD ["uv", "run", "--no-dev", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

生产环境通常还会在前面加一层 Nginx 做反向代理、HTTPS 和静态资源服务。

部署上线前建议检查：

关闭 --reload，按需设置 --workers；
数据库密码、密钥等敏感配置从环境变量读取，不要提交到仓库；
CORS 只放行实际使用的前端域名；
不需要对外暴露接口文档时，可以用 FastAPI(docs_url=None, redoc_url=None) 关闭 /docs 和 /redoc；
上线前执行 uv run aerich upgrade 同步数据库结构。

总结

主题	核心 API
路由	`@app.get/post/put/patch/delete`、`APIRouter`、`include_router`
参数	路径参数、查询参数、`Path`、`Query`、`Header`、`Cookie`
请求体	Pydantic `BaseModel`、`Field`、`field_validator`、`Form`、`UploadFile`
响应	返回类型注解、`response_model`、`status_code`、各种 `Response` 子类
异常	`HTTPException`、`@app.exception_handler`
依赖注入	`Depends`、`Annotated`、`yield` 依赖、`dependencies=[...]`
中间件	`@app.middleware("http")`、`CORSMiddleware`
生命周期	`lifespan`、`BackgroundTasks`
数据库	Tortoise-ORM、`RegisterTortoise`、Aerich 迁移
测试与部署	`TestClient`、`pytest`、多 worker、Docker

OAuth2 + JWT 认证、WebSocket、SQLModel/SQLAlchemy 等内容本文没有涉及，可以参考 FastAPI 官方文档。