All writing Notes & observations

FastAPI 实战教程:面向 Python 开发者的现代 API 入门

面向已有 Python 基础的开发者,从第一个接口开始掌握 FastAPI 的路由、Pydantic 校验、依赖注入、异步 I/O、测试和部署。

FastAPI 实战教程:面向 Python 开发者的现代 API 入门

如果你已经会 Python 函数、类型提示、虚拟环境和一点 HTTP 基础,FastAPI 会给你一种很顺手的体验:你写普通函数和 Pydantic 模型,它负责把请求解析、参数校验、JSON 序列化和 OpenAPI 文档串起来。

FastAPI 实战教程封面

这篇文章不从“什么是 Web 框架”讲起,而是面向已经有 Python 基础的开发者,直接搭一个可运行的小 API,再逐步补上工程里绕不开的几件事:数据校验、错误处理、依赖注入、异步选择、项目拆分、测试和部署。

你已有的 Python 能力 在 FastAPI 中会怎样复用
函数与装饰器 一个路径函数就是一个接口处理函数
类型提示 自动解析路径参数、查询参数和请求体
类与数据模型 用 Pydantic 定义请求和响应契约
异常处理 用 HTTPException 表达 HTTP 错误
模块拆分 用 APIRouter 管理多组接口
测试习惯 用 TestClient 写接口级单元测试

一、FastAPI 解决的核心问题

FastAPI 的价值不是“少写几行 Flask 代码”,而是把现代 API 后端中高频出现的一组能力整合在一起:

  • 基于类型提示的参数解析;
  • 基于 Pydantic 的请求体和响应体校验;
  • 自动生成 OpenAPI、Swagger UI 和 ReDoc;
  • 支持同步函数和异步函数;
  • 通过依赖注入组织鉴权、数据库会话、配置读取等横切逻辑;
  • 基于 ASGI,天然适合高并发 I/O 场景。

FastAPI 请求生命周期

一次请求进入 FastAPI 后,通常会经过这样的链路:

  1. Uvicorn 这类 ASGI Server 接收 HTTP 请求;
  2. FastAPI 根据 method 和 path 找到路径函数;
  3. 解析 path、query、header、cookie、body 等输入;
  4. 用 Pydantic 做类型转换和约束校验;
  5. 执行业务函数;
  6. 根据 response_model 输出稳定 JSON 结构;
  7. 同步更新 OpenAPI 文档。

也就是说,FastAPI 希望你把“接口契约”尽量写进 Python 类型和模型里,而不是散落在注释、Wiki 和前端口头约定中。


二、环境准备

截至本文写作时,FastAPI 官方文档推荐安装标准依赖包:

1
pip install "fastapi[standard]"

建议从一个干净虚拟环境开始:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
mkdir fastapi-shop
cd fastapi-shop

python -m venv .venv

# Windows PowerShell
.\.venv\Scripts\Activate.ps1

# macOS / Linux
source .venv/bin/activate

pip install "fastapi[standard]"

其中 fastapi[standard] 会带上常用运行和开发能力,比如 ASGI Server、交互式文档相关依赖等。如果你在生产项目中追求更小依赖面,也可以研究 fastapi-slim,但入门和多数业务项目建议先用标准安装。


三、第一个可运行 API

先写一个单文件版本,文件名为 main.py:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
from typing import Annotated
from uuid import UUID, uuid4

from fastapi import Depends, FastAPI, Header, HTTPException, Query, status
from pydantic import BaseModel, Field


app = FastAPI(
    title="FastAPI Shop",
    version="0.1.0",
    description="A tiny shop API built with FastAPI.",
)


class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=80, examples=["Keyboard"])
    price: float = Field(..., gt=0, examples=[299.0])
    tags: list[str] = Field(default_factory=list, max_length=10)


class ItemUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=1, max_length=80)
    price: float | None = Field(default=None, gt=0)
    tags: list[str] | None = Field(default=None, max_length=10)


class ItemRead(BaseModel):
    id: UUID
    name: str
    price: float
    tags: list[str]


items: dict[UUID, ItemRead] = {}


def require_token(
    x_token: Annotated[str | None, Header(alias="X-Token")] = None,
) -> str:
    if x_token != "dev-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
        )
    return x_token


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/items", response_model=list[ItemRead])
def list_items(
    token: Annotated[str, Depends(require_token)],
    q: Annotated[str | None, Query(min_length=1, max_length=30)] = None,
) -> list[ItemRead]:
    result = list(items.values())

    if q is None:
        return result

    keyword = q.lower()
    return [item for item in result if keyword in item.name.lower()]


@app.post("/items", response_model=ItemRead, status_code=status.HTTP_201_CREATED)
def create_item(
    data: ItemCreate,
    token: Annotated[str, Depends(require_token)],
) -> ItemRead:
    item = ItemRead(id=uuid4(), **data.model_dump())
    items[item.id] = item
    return item


@app.get("/items/{item_id}", response_model=ItemRead)
def read_item(
    item_id: UUID,
    token: Annotated[str, Depends(require_token)],
) -> ItemRead:
    try:
        return items[item_id]
    except KeyError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Item not found",
        )


@app.patch("/items/{item_id}", response_model=ItemRead)
def update_item(
    item_id: UUID,
    data: ItemUpdate,
    token: Annotated[str, Depends(require_token)],
) -> ItemRead:
    item = read_item(item_id, token)
    updates = data.model_dump(exclude_unset=True)
    updates = {key: value for key, value in updates.items() if value is not None}
    updated = item.model_copy(update=updates)
    items[item_id] = updated
    return updated


@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(
    item_id: UUID,
    token: Annotated[str, Depends(require_token)],
) -> None:
    if item_id not in items:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Item not found",
        )
    del items[item_id]

启动开发服务器:

1
fastapi dev main.py

如果你的环境没有 FastAPI CLI,也可以用 Uvicorn:

1
uvicorn main:app --reload

打开:

  • http://127.0.0.1:8000/health
  • http://127.0.0.1:8000/docs
  • http://127.0.0.1:8000/redoc
  • http://127.0.0.1:8000/openapi.json

试着创建一个商品:

1
2
3
4
curl -X POST "http://127.0.0.1:8000/items" \
  -H "Content-Type: application/json" \
  -H "X-Token: dev-token" \
  -d '{"name":"Keyboard","price":299,"tags":["peripheral"]}'

如果你把 price 改成负数,FastAPI 会直接返回 422 Unprocessable Entity,并告诉你哪个字段不符合约束。你不需要在业务函数里手写一堆 if price <= 0。


四、类型提示如何变成接口契约

FastAPI 里有三类输入最常见:

输入位置 示例 FastAPI 如何识别
路径参数 /items/{item_id} 函数参数名与路径模板一致
查询参数 /items?q=key 函数参数不在路径里,且不是 Pydantic body
请求体 JSON body 参数类型继承自 BaseModel

例如:

1
2
3
4
5
6
@app.get("/items/{item_id}")
def read_item(
    item_id: UUID,
    q: Annotated[str | None, Query(min_length=1)] = None,
) -> dict[str, str]:
    return {"item_id": str(item_id), "q": q or ""}

这里 item_id: UUID 会让 FastAPI 自动把字符串转换为 UUID 对象;如果传入的路径不是合法 UUID,请求会在进入函数前失败。

Annotated 的好处是把“类型”和“接口约束”分开:

1
q: Annotated[str | None, Query(min_length=1, max_length=30)] = None

读作:q 的业务类型是 str | None,同时它作为查询参数时必须满足长度约束。这样写比把校验逻辑塞进函数体更清晰,也能同步生成文档。


五、Pydantic:把边界数据变可信

Pydantic 是 FastAPI 的数据契约核心。请求进入系统时,外部世界给你的都是“不可信字符串或 JSON”;经过 Pydantic 模型之后,业务函数拿到的是已经转换和校验过的 Python 对象。

Pydantic 校验与响应模型

请求模型和响应模型建议分开写:

1
2
3
4
5
6
7
8
class UserCreate(BaseModel):
    email: str
    password: str


class UserRead(BaseModel):
    id: UUID
    email: str

这样做有两个直接收益:

  1. password 可以出现在创建请求里,但不会从响应里泄漏出去;
  2. 数据库模型、请求模型、响应模型不再互相污染。

response_model 还有一个经常被低估的价值:它能过滤掉你不想暴露的字段。

1
2
3
4
5
6
7
@app.post("/users", response_model=UserRead)
def create_user(data: UserCreate) -> dict[str, object]:
    return {
        "id": uuid4(),
        "email": data.email,
        "password_hash": "internal-only",
    }

即使函数返回值里有 password_hash,最终响应也只会保留 UserRead 声明的字段。对于 API 稳定性和安全边界,这一点很重要。


六、依赖注入:把横切逻辑拿出接口函数

开发业务 API 时,很多逻辑并不属于某一个接口:

  • 读取当前登录用户;
  • 校验 Token;
  • 创建数据库 Session;
  • 读取配置;
  • 记录审计日志;
  • 检查权限。

FastAPI 用 Depends 处理这类逻辑。前面的 require_token 就是一个最小例子:

1
2
3
4
5
6
def require_token(
    x_token: Annotated[str | None, Header(alias="X-Token")] = None,
) -> str:
    if x_token != "dev-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return x_token

在接口中声明依赖:

1
2
3
@app.get("/items")
def list_items(token: Annotated[str, Depends(require_token)]) -> list[ItemRead]:
    return list(items.values())

这样接口函数会在执行前先运行 require_token。如果依赖抛出异常,业务函数不会继续执行。

依赖也可以继续依赖其他依赖:

1
2
3
4
5
6
7
8
def get_settings() -> Settings:
    return Settings()


def get_repository(
    settings: Annotated[Settings, Depends(get_settings)],
) -> ItemRepository:
    return ItemRepository(settings.database_url)

这种组合方式让 FastAPI 项目很适合做“薄接口层 + 明确业务层”的结构。


七、项目变大后怎么拆

单文件很适合入门,但真实项目通常会快速长出用户、订单、支付、后台管理、文件上传等资源。推荐按职责拆分,而不是按“今天写了哪个接口”随手堆文件。

FastAPI 推荐项目结构

一个常见结构是:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
fastapi-shop/
├── app/
│   ├── main.py
│   ├── api/
│   │   └── routers/
│   │       ├── items.py
│   │       └── users.py
│   ├── core/
│   │   └── config.py
│   ├── schemas.py
│   ├── models.py
│   ├── services.py
│   └── db.py
├── tests/
│   └── test_items.py
└── requirements.txt

app/api/routers/items.py 可以这样写:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from typing import Annotated
from uuid import UUID

from fastapi import APIRouter, Depends

from app.schemas import ItemCreate, ItemRead
from app.services import ItemService, get_item_service


router = APIRouter(prefix="/items", tags=["items"])


@router.post("", response_model=ItemRead, status_code=201)
def create_item(
    data: ItemCreate,
    service: Annotated[ItemService, Depends(get_item_service)],
) -> ItemRead:
    return service.create(data)


@router.get("/{item_id}", response_model=ItemRead)
def read_item(
    item_id: UUID,
    service: Annotated[ItemService, Depends(get_item_service)],
) -> ItemRead:
    return service.get(item_id)

app/main.py 只负责组装:

1
2
3
4
5
6
7
from fastapi import FastAPI

from app.api.routers import items


app = FastAPI(title="FastAPI Shop")
app.include_router(items.router)

这里有一个判断标准:如果路由函数开始出现复杂分支、数据库查询细节、外部 API 调用和权限规则,说明它已经不只是“HTTP 编排”了,应该把逻辑移到 service 或 repository 层。


八、async def 不等于自动更快

FastAPI 支持两种路径函数:

1
2
3
4
5
6
7
8
@app.get("/sync")
def sync_handler() -> dict[str, str]:
    return {"mode": "sync"}


@app.get("/async")
async def async_handler() -> dict[str, str]:
    return {"mode": "async"}

什么时候用 async def?先看函数内部在做什么。

FastAPI 异步 I/O 使用判断

可以用这张表快速判断:

场景 建议
调用 async 数据库驱动,如 async SQLAlchemy、asyncpg 用 async def
调用 async HTTP 客户端,如 httpx.AsyncClient 用 async def
调用普通同步库,如 requests、同步数据库驱动 用 def,或放到线程池/任务队列
做大量 CPU 计算,如图像处理、模型推理 不要指望 async 提速,考虑进程池或后台任务
不确定库是否异步 先按同步写,确认后再改

最常见的坑是:写了 async def,但函数内部直接调用 requests.get() 或同步数据库驱动。这会阻塞事件循环,让其他请求也跟着等待。

如果需要在接口中调用外部 HTTP 服务,可以这样:

1
2
3
4
5
6
7
8
9
import httpx


@app.get("/exchange-rate")
async def get_exchange_rate() -> dict[str, float]:
    async with httpx.AsyncClient(timeout=5) as client:
        response = await client.get("https://api.example.com/rate")
        response.raise_for_status()
    return response.json()

注意:异步不是语法装饰,它是一种 I/O 调度模型。只有你的数据库、HTTP 客户端、缓存客户端等调用链也支持 async,收益才会明显。


九、错误处理与状态码

HTTP API 不应该只返回 200,然后把错误塞进 JSON 里的 code 字段。状态码本身就是协议的一部分。

常用状态码可以这样理解:

状态码 适用场景
200 OK 查询、更新成功并返回内容
201 Created 创建资源成功
204 No Content 删除成功且不返回 body
400 Bad Request 请求语义不合法
401 Unauthorized 没有有效身份凭证
403 Forbidden 已登录但没有权限
404 Not Found 资源不存在
409 Conflict 资源状态冲突,如重复创建
422 Unprocessable Entity 参数类型或字段约束不通过

FastAPI 中可以用 HTTPException 表达业务错误:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from fastapi import HTTPException, status


def get_item_or_404(item_id: UUID) -> ItemRead:
    item = items.get(item_id)
    if item is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Item not found",
        )
    return item

如果项目中错误结构需要统一,比如都返回 {"code": "...", "message": "..."},可以再引入全局异常处理器。但在入门阶段,先把正确状态码用好,比过早设计复杂错误体系更重要。


十、测试:用接口行为保护重构

FastAPI 提供 TestClient,可以像调用普通 HTTP 接口一样测试你的应用。

安装测试依赖:

1
pip install pytest

tests/test_items.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from fastapi.testclient import TestClient

from main import app


client = TestClient(app)


def test_create_item() -> None:
    response = client.post(
        "/items",
        headers={"X-Token": "dev-token"},
        json={"name": "Mouse", "price": 99.0, "tags": ["peripheral"]},
    )

    assert response.status_code == 201
    body = response.json()
    assert body["name"] == "Mouse"
    assert body["price"] == 99.0
    assert "id" in body


def test_reject_invalid_price() -> None:
    response = client.post(
        "/items",
        headers={"X-Token": "dev-token"},
        json={"name": "Mouse", "price": -1},
    )

    assert response.status_code == 422


def test_reject_missing_token() -> None:
    response = client.get("/items")

    assert response.status_code == 401

运行:

1
pytest

接口测试的重点不是追求行覆盖率,而是锁住 API 行为:状态码、响应字段、错误场景和鉴权路径。只要这些行为被测试保护,后面把内存字典替换成数据库时就安心很多。


十一、部署前需要想清楚的事

本地开发时可以用:

1
fastapi dev main.py

生产环境不要直接用开发命令。你可以使用 FastAPI CLI 的生产命令:

1
fastapi run app/main.py --host 0.0.0.0 --port 8000

也可以显式使用 Uvicorn:

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

FastAPI 从本地到生产的部署路径

一个最小 Dockerfile 可以长这样:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app ./app

EXPOSE 8000
CMD ["fastapi", "run", "app/main.py", "--host", "0.0.0.0", "--port", "8000"]

生产环境至少要补齐这些点:

项目 为什么重要
固定依赖版本 避免一次重新构建引入不可预期变更
环境变量管理 数据库地址、密钥、第三方 Token 不要写死在代码里
健康检查 让负载均衡和容器平台知道服务是否可用
结构化日志 方便按 request_id、user_id、status_code 排查问题
数据库迁移 用 Alembic 等工具管理 schema 变化
超时与重试 防止外部服务拖垮自己的接口
限流与鉴权 公开 API 必须考虑滥用和越权
HTTPS 与反向代理 生产入口通常交给 Nginx、Traefik、云负载均衡等

十二、常见坑

坑 更稳的做法
把所有代码写进 main.py 入门可以,变大后用 APIRouter、service、db 分层
路由函数里写大量业务分支 路由层只处理 HTTP,业务规则移到 service
不写 response_model 为公开 API 写响应模型,保护字段边界
async def 里调用同步阻塞库 同步库用 def,异步链路才用 async def
返回 200 表示所有结果 正确使用 201、204、401、404、409、422
把密钥写进代码 用环境变量或 Secret 管理
不写测试就接数据库 先用行为测试保护接口,再替换持久化实现

十三、学习路线

到这里,你已经掌握了 FastAPI 最重要的主干。接下来可以按这个顺序继续:

  1. SQLAlchemy 2.x 或 SQLModel:把内存字典换成数据库;
  2. Alembic:管理数据库迁移;
  3. OAuth2 / JWT:实现真实登录鉴权;
  4. BackgroundTasks / Celery / RQ:处理慢任务;
  5. 中间件:统一日志、request_id、CORS、耗时统计;
  6. OpenTelemetry:接入链路追踪;
  7. Docker Compose / Kubernetes:把 API、数据库、缓存组成可部署系统。

FastAPI 的学习曲线很友好,但它不是“玩具框架”。当你把类型契约、依赖注入、测试和部署纪律一起用起来,它完全可以支撑严肃的生产 API。


参考资料