一命速通 FastAPI 的文章封面
返回文章列表
Neuroblue writing

一命速通 FastAPI

从零搭一个可维护的 FastAPI 后端:路由、参数、Pydantic、依赖注入、数据库、测试、鉴权和部署。

FastAPI 是一个用 Python 写现代 Web API 的框架。它的爽点很直接:写类型标注,框架帮你做参数解析、数据校验、OpenAPI 文档生成和交互式调试页面。对想快速做 AI 工具后端、自动化服务、管理后台 API、爬虫任务接口的人来说,它是很适合一命速通的技术栈。

本文目标不是背完文档,而是建立一条能动手的路线:先跑起来,再理解请求如何进入函数,最后把项目整理成可维护的后端服务。

0. 你需要先知道什么

FastAPI 的几个关键词:

  • path operation:路由函数,也就是 GET /items/{id} 这种接口。
  • Pydantic:数据模型与校验工具,用来定义请求体、响应体、配置等结构。
  • type hints:Python 类型标注,FastAPI 会用它理解参数和返回值。
  • dependency injection:依赖注入,用来复用数据库会话、鉴权、分页参数等公共逻辑。
  • OpenAPI:自动生成的 API 描述标准,FastAPI 默认会生成文档。
  • ASGI:Python 异步 Web 服务接口,常用运行器是 Uvicorn。

1. 安装与启动

创建环境:

mkdir fastapi-one-life
cd fastapi-one-life
python -m venv .venv
.\.venv\Scripts\Activate.ps1

安装:

pip install "fastapi[standard]"

新建 main.py

from fastapi import FastAPI

app = FastAPI(title="One Life FastAPI")


@app.get("/")
def read_root():
    return {"message": "hello fastapi"}

启动:

fastapi dev main.py

默认可以访问:

http://127.0.0.1:8000/
http://127.0.0.1:8000/docs
http://127.0.0.1:8000/redoc

/docs 是 Swagger UI,适合边写边调接口。

2. 路由:先把请求接住

最基本的接口:

from fastapi import FastAPI

app = FastAPI()


@app.get("/health")
def health_check():
    return {"status": "ok"}

常见 HTTP 方法:

@app.get("/items")
def list_items():
    return []


@app.post("/items")
def create_item():
    return {"created": True}


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


@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    return {"deleted": item_id}

路由设计建议:

  • 列表:GET /items
  • 详情:GET /items/{item_id}
  • 创建:POST /items
  • 更新:PUT /items/{item_id}PATCH /items/{item_id}
  • 删除:DELETE /items/{item_id}

3. 路径参数与查询参数

路径参数来自 URL 的一部分:

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

访问:

GET /users/1

FastAPI 会把 "1" 转成 int。如果传入 abc,它会自动返回校验错误。

查询参数来自 ? 后面:

@app.get("/posts")
def list_posts(page: int = 1, size: int = 10, keyword: str | None = None):
    return {
        "page": page,
        "size": size,
        "keyword": keyword,
    }

访问:

GET /posts?page=2&size=20&keyword=fastapi

4. 请求体:用 Pydantic 定义数据形状

创建文章时,客户端会发 JSON。用 Pydantic 模型接收:

from pydantic import BaseModel, Field


class PostCreate(BaseModel):
    title: str = Field(min_length=1, max_length=80)
    content: str
    published: bool = False


@app.post("/posts")
def create_post(payload: PostCreate):
    return {
        "title": payload.title,
        "content": payload.content,
        "published": payload.published,
    }

请求:

{
  "title": "一命速通 FastAPI",
  "content": "正文内容",
  "published": true
}

如果少字段、字段类型错、标题为空,FastAPI 会自动返回错误响应。你不用手写一堆 if not title

5. 响应模型:不要把不该返回的字段吐出去

后端经常有内部字段,比如密码哈希、删除标记、权限字段。响应模型用来控制输出:

from pydantic import BaseModel


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


@app.get("/users/{user_id}", response_model=UserOut)
def get_user(user_id: int):
    return {
        "id": user_id,
        "username": "neuroblue",
        "password_hash": "do-not-return",
    }

实际响应只会包含:

{
  "id": 1,
  "username": "neuroblue"
}

6. 依赖注入:把公共逻辑抽出去

依赖注入适合处理:

  • 数据库连接
  • 当前用户
  • 分页参数
  • 权限校验
  • 配置读取

例子:统一分页参数。

from fastapi import Depends, Query
from pydantic import BaseModel


class Pagination(BaseModel):
    page: int
    size: int


def get_pagination(
    page: int = Query(default=1, ge=1),
    size: int = Query(default=10, ge=1, le=100),
) -> Pagination:
    return Pagination(page=page, size=size)


@app.get("/posts")
def list_posts(pagination: Pagination = Depends(get_pagination)):
    return pagination

这样每个列表接口都可以复用同一套分页规则。

7. 错误处理:该失败就优雅失败

抛出 HTTP 错误:

from fastapi import HTTPException


@app.get("/posts/{post_id}")
def get_post(post_id: int):
    if post_id != 1:
        raise HTTPException(status_code=404, detail="Post not found")
    return {"id": post_id, "title": "FastAPI"}

常用状态码:

  • 400:请求有问题。
  • 401:未登录。
  • 403:无权限。
  • 404:资源不存在。
  • 409:状态冲突,比如用户名重复。
  • 422:FastAPI 默认的数据校验失败。
  • 500:服务端错误。

8. 项目结构:别把所有东西塞进 main.py

小 demo 可以只有 main.py。稍微认真一点,建议这样拆:

app/
  main.py
  core/
    config.py
  api/
    routes/
      posts.py
      users.py
  schemas/
    post.py
    user.py
  services/
    post_service.py
  repositories/
    post_repository.py
  db/
    session.py
    models.py
tests/
  test_posts.py

分层思路:

  • api/routes:只处理 HTTP 输入输出。
  • schemas:Pydantic 请求体和响应体。
  • services:业务逻辑。
  • repositories:数据库读写。
  • db:数据库连接、ORM 模型。
  • core:配置、日志、安全相关工具。

9. 实战:Todo API

下面做一个内存版 Todo API。它不是最终生产形态,但可以让你把路由、模型、错误处理、依赖注入串起来。

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

app = FastAPI(title="Todo API")


class TodoCreate(BaseModel):
    title: str = Field(min_length=1, max_length=80)
    done: bool = False


class TodoUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=80)
    done: bool | None = None


class TodoOut(BaseModel):
    id: int
    title: str
    done: bool


class PageParams(BaseModel):
    page: int
    size: int


def get_page_params(
    page: int = Query(default=1, ge=1),
    size: int = Query(default=10, ge=1, le=100),
) -> PageParams:
    return PageParams(page=page, size=size)


todos: dict[int, TodoOut] = {}
next_id = 1


@app.post("/todos", response_model=TodoOut)
def create_todo(payload: TodoCreate):
    global next_id
    todo = TodoOut(id=next_id, title=payload.title, done=payload.done)
    todos[next_id] = todo
    next_id += 1
    return todo


@app.get("/todos", response_model=list[TodoOut])
def list_todos(page_params: PageParams = Depends(get_page_params)):
    start = (page_params.page - 1) * page_params.size
    end = start + page_params.size
    return list(todos.values())[start:end]


@app.get("/todos/{todo_id}", response_model=TodoOut)
def get_todo(todo_id: int):
    todo = todos.get(todo_id)
    if todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo


@app.patch("/todos/{todo_id}", response_model=TodoOut)
def update_todo(todo_id: int, payload: TodoUpdate):
    todo = todos.get(todo_id)
    if todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")

    updated = todo.model_copy(
        update={
            "title": payload.title if payload.title is not None else todo.title,
            "done": payload.done if payload.done is not None else todo.done,
        }
    )
    todos[todo_id] = updated
    return updated


@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
    if todo_id not in todos:
        raise HTTPException(status_code=404, detail="Todo not found")
    del todos[todo_id]
    return {"deleted": todo_id}

测试一下:

fastapi dev main.py

打开:

http://127.0.0.1:8000/docs

在文档里依次调用:

  1. POST /todos
  2. GET /todos
  3. PATCH /todos/{todo_id}
  4. DELETE /todos/{todo_id}

这就是 FastAPI 最小闭环。

10. 接数据库:从内存走向真实服务

真实项目不能用内存字典,服务重启数据就没了。常见组合:

  • PostgreSQL + SQLAlchemy
  • SQLite + SQLModel
  • MySQL + SQLAlchemy
  • MongoDB + Motor 或 Beanie

学习阶段可以先用 SQLite。核心思路:

  1. 定义数据库模型。
  2. 创建数据库连接。
  3. 每个请求获取一个 session。
  4. 在 repository 或 service 里操作数据库。
  5. 用 Pydantic schema 控制输入输出。

数据库依赖示意:

from collections.abc import Generator


def get_session() -> Generator:
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

路由中使用:

@app.get("/todos")
def list_todos(session = Depends(get_session)):
    ...

这里不要急着复制大型模板。先把“每个请求一个 session,用完关闭”这件事理解清楚。

11. 鉴权:先理解 Bearer Token

FastAPI 支持 OAuth2、JWT、API Key 等方式。学习阶段可以先理解这个流程:

  1. 用户提交用户名密码。
  2. 服务端校验用户。
  3. 服务端签发 access token。
  4. 客户端请求时带上 Authorization: Bearer <token>
  5. 后端解析 token,得到当前用户。
  6. 路由通过依赖注入拿到 current_user

伪代码:

from fastapi import Depends


def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)
    if user is None:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user


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

不要在学习阶段一上来就把 JWT、刷新 token、权限系统、RBAC 全部堆满。先把“当前用户如何进入路由函数”跑通。

12. 测试:别只靠手点 docs

安装:

pip install pytest httpx

测试示例:

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)


def test_create_todo():
    response = client.post("/todos", json={"title": "learn fastapi"})
    assert response.status_code == 200
    data = response.json()
    assert data["title"] == "learn fastapi"
    assert data["done"] is False

运行:

pytest

建议至少测:

  • 创建成功。
  • 参数缺失。
  • 参数类型错误。
  • 查询不存在资源。
  • 更新资源。
  • 删除资源。
  • 鉴权失败。

13. 部署:开发命令和生产命令分开

开发时:

fastapi dev main.py

生产时通常使用 Uvicorn 或容器部署:

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

生产前检查:

  • 是否关闭 debug。
  • 是否设置环境变量。
  • 数据库连接是否来自配置。
  • CORS 是否限制来源。
  • 日志是否可观察。
  • 是否有健康检查接口。
  • 是否有错误监控。
  • 是否有数据库迁移方案。

14. CORS:前后端分离必须处理

前端和后端域名不同,就会遇到 CORS。

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

生产环境不要直接放开所有来源:

allow_origins=["*"]

除非你明确知道自己在做公共 API。

15. 学习路线

一命速通路线:

  1. 跑通 hello world
  2. 学会路径参数、查询参数、请求体。
  3. 用 Pydantic 写请求和响应模型。
  4. 写 Todo API。
  5. 加分页、错误处理、CORS。
  6. 拆项目结构。
  7. 接 SQLite 或 PostgreSQL。
  8. 加测试。
  9. 加登录鉴权。
  10. 打包部署。

不要一开始就追求“大而全”。后端能力是从一个个可运行的接口里长出来的。

16. 常见坑

16.1 async 到底要不要写

如果函数里主要是普通同步代码,可以写:

def read_items():
    ...

如果使用异步库,比如异步数据库客户端、异步 HTTP 客户端,可以写:

async def read_items():
    ...

不要为了看起来高级,把所有函数都写成 async

16.2 返回 ORM 对象为什么报错

通常是响应模型和 ORM 对象没有对齐,或者没有正确配置模型转换。简单原则:路由返回前,确认输出结构符合 response_model

16.3 为什么 422

422 基本是参数校验失败。去看响应体里的 detail,它会告诉你哪个字段错了。

16.4 CORS 明明配了还是不行

检查:

  • 前端请求地址是否正确。
  • 是否带了 credentials。
  • allow_origins 是否包含前端地址。
  • 浏览器实际报错的是预检请求还是业务请求。

16.5 模型怎么拆

常见拆法:

  • UserCreate:创建用户的请求体。
  • UserUpdate:更新用户的请求体。
  • UserOut:返回给前端的结构。
  • UserInDB:数据库内部结构。

不要一个模型从头用到尾。

17. 最小实战作业

如果你真的想速通,做这个项目:

FastAPI 个人任务系统

要求:

  • 用户注册和登录。
  • JWT 鉴权。
  • Todo CRUD。
  • Todo 分类。
  • 分页查询。
  • 关键词搜索。
  • SQLite 或 PostgreSQL。
  • Pytest 覆盖核心接口。
  • Docker 部署。

完成后你就已经跨过 FastAPI 的第一道门了。剩下的就是工程习惯和项目经验。

18. 参考资料