fuchencong

The way I am

0%

full-stack-fastapi-template 源码分析

发表于 分类于

这篇文章将分析 full-stack-fastapi-template 的源码，学习使用 fastapi 框架构建 Web 应用时应该如何组织代码结构。这篇文章主要分析其后端部分实现，但首先我们仍然会对该项目做一个整体说明。

项目整体说明

Full Stack FastAPI Template 是一个生产级全栈 Web 应用模板，旨在为开发者提供一个开箱即用的现代化 Web 应用脚手架。它解决了从零开始搭建全栈应用时的重复性配置工作，让开发者能够快速聚焦于业务逻辑开发。

它主要包含了如下核心功能：

  • 用户认证与授权系统：完整的用户身份验证和权限管理机制
  • 用户管理模块：完整的用户 CRUD 操作，支持普通用户和管理员两种角色
  • Items 业务模块：示例业务实体，展示典型的 CRUD 操作和数据所有权控制

整个项目包含前端和后端这两个核心部分，前端是一个现代化的 React 应用，使用自动生成 TypeScript 客户端来与后端 API 交互。同时项目使用了现代化的基础设施，如通过 Docker 使应用容器化，使用 Docker Compose 编排多个应用容器，使用 Traefik 进行反向代理，使用 Playwright 进行 E2E（端到端）测试等等。

项目的整体代码结构组织如下：

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
full-stack-fastapi-template/
├── .github/                    # GitHub 工作流配置
│   └── workflows/              # CI/CD Pipeline
├── .copier/                    # Copier 模板配置
│   └── update_dotenv.py        # 项目生成后脚本
├── backend/                    # 后端应用
│   ├── app/                    # 应用核心代码
│   │   ├── alembic/            # 数据库迁移
│   │   │   ├── env.py          # Alembic 环境配置
│   │   │   └── versions/       # 迁移版本文件
│   │   ├── api/                # API 层
│   │   │   ├── routes/         # 路由处理器
│   │   │   ├── deps.py         # 依赖注入
│   │   │   └── main.py         # 路由聚合
│   │   ├── core/               # 核心配置
│   │   │   ├── config.py       # 应用配置
│   │   │   ├── db.py           # 数据库连接
│   │   │   └── security.py     # 安全模块
│   │   ├── email-templates/    # 邮件模板
│   │   ├── __init__.py
│   │   ├── crud.py             # CRUD 操作
│   │   ├── initial_data.py     # 初始数据
│   │   ├── main.py             # 应用入口
│   │   ├── models.py           # 数据模型
│   │   └── utils.py            # 工具函数
│   ├── tests/                  # 测试代码
│   │   ├── api/routes/         # API 测试
│   │   ├── crud/               # CRUD 测试
│   │   ├── scripts/            # 脚本测试
│   │   ├── utils/              # 测试工具
│   │   └── conftest.py         # Pytest 配置
│   ├── Dockerfile              # 后端镜像构建
│   └── pyproject.toml          # Python 项目配置
├── frontend/                   # 前端应用
│   ├── src/                    # 源代码
│   │   ├── client/             # 自动生成的 API 客户端
│   │   ├── components/         # React 组件
│   │   │   ├── Admin/          # 管理员相关组件
│   │   │   ├── Common/         # 通用组件
│   │   │   ├── Items/          # Items 相关组件
│   │   │   ├── Sidebar/        # 侧边栏组件
│   │   │   ├── UserSettings/   # 用户设置组件
│   │   │   └── ui/             # shadcn/ui 组件库
│   │   ├── hooks/              # 自定义 Hooks
│   │   ├── lib/                # 工具库
│   │   ├── routes/             # 路由页面
│   │   ├── index.css           # 全局样式
│   │   ├── main.tsx            # 应用入口
│   │   └── utils.ts            # 工具函数
│   ├── tests/                  # E2E 测试
│   ├── public/                 # 静态资源
│   ├── Dockerfile              # 前端镜像构建
│   ├── openapi-ts.config.ts    # OpenAPI 客户端生成配置
│   ├── package.json            # NPM 配置
│   └── vite.config.ts          # Vite 配置
├── hooks/                      # Copier 钩子脚本
├── img/                        # 文档图片
├── scripts/                    # 部署脚本
│   └── prestart.sh             # 容器启动前脚本
├── .env                        # 环境变量
├── compose.yml                 # Docker Compose 生产配置
├── compose.override.yml        # Docker Compose 开发覆盖
├── compose.traefik.yml         # Traefik 配置
├── copier.yml                  # Copier 模板定义
├── deployment.md               # 部署文档
├── development.md              # 开发文档
└── README.md                   # 项目说明

接下来我们重点学习其后端代码实现，学习 fastapi 框架构建 Web 应用后端的一些基本方法。

后端代码详解

程序主入口

程序的主入口 backend/app/main.py 定义了全局的 FastAPI app 实例：

1
2
3
4
5
6
7
app = FastAPI(
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.API_V1_STR}/openapi.json",
    generate_unique_id_function=custom_generate_unique_id,
)

app.include_router(api_router, prefix=settings.API_V1_STR)
  • 通过 FastAPI() 创建一个 FastAPI 应用实例，并通过 include_router 挂载所有的业务路由
  • 默认的路由前缀为 /api/v1

main.py 中除了定义 FastAPI app 实例，还进行了一些全局初始化操作：

  • 初始化 sentry_sdk
1
2
if settings.SENTRY_DSN and settings.ENVIRONMENT != "local":
    sentry_sdk.init(dsn=str(settings.SENTRY_DSN), enable_tracing=True)
  • 设置 CORS 策略

那具体如何启动这个 FastAPI 应用呢？答案在 backend/Dockerfile 中，程序的启动命令是：

1
CMD ["fastapi", "run", "--workers", "4", "app/main.py"]

fastapi 这是 FastAPI 官方提供的命令行工具。它会自动检测环境（如果是生产环境，它会默认优化性能）。它其实底层调用了像 uvicorn 或 gunicorn 这样的服务器软件。

配置管理

后端应用一般都有自己的配置，包括如何连接数据库、如何与外部服务交互等等。在 full-stack-fastapi-template 中，配置管理的实现位于 backend/app/core/config.py 文件中。它使用了 pydantic、pydantic-settings 等库来完成全局配置的加载、验证、管理。

BaseSettings 是 pydantic-settings 库的核心类，用于从环境变量、配置文件（如 .env）、命令行参数等来源加载和管理应用配置，同时提供类型验证、默认值处理、嵌套配置等强大功能。应用程序通过定义继承 BaseSettings 的自定义类，来管理各种配置项。

  • 会自动从环境变量、.env 文件、命令行参数等来源读取字段值
  • 支持类型注解（如 str、int、list、嵌套 BaseModel 等），自动验证字段类型合法性
  • 未通过外部来源设置的字段，将使用代码中定义的默认值
  • 可以通过 model_config（SettingsConfigDict 类型）配置解析规则与行为
1
2
3
4
5
6
7
8
9
10
11
class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        # Use top level .env file (one level above ./backend/)
        env_file="../.env",
        env_ignore_empty=True,
        extra="ignore",
    )
    API_V1_STR: str = "/api/v1"
    SECRET_KEY: str = secrets.token_urlsafe(32)

settings = Settings()  # type: ignore

这里是定义了一个全局的配置对象 settings，包括了应用程序的所有配置，后续其他模块需要使用配置时，直接通过 from app.core.config import settings 只用这个全局配置对象即可。

API 路由

backend/app/api 是 FastAPI 项目的 API 层核心，承担 HTTP 请求路由、依赖注入、权限控制等职责。该模块采用分层架构 + 依赖注入的设计，实现了高度解耦的代码组织。backend/app/api/main.py 作为路由聚合器，统一注册所有子路由：

1
2
3
4
api_router = APIRouter()
api_router.include_router(login.router)
api_router.include_router(users.router)
......

通过 FastAPI 的依赖注入机制，实现代码解耦和复用：通过 Annotated 封装可复用的依赖项，无需重复编写认证、数据库连接、Token 解析等通用逻辑：

1
2
3
4
5
6
7
SessionDep = Annotated[Session, Depends(get_db)]
TokenDep = Annotated[str, Depends(reusable_oauth2)]
CurrentUser = Annotated[User, Depends(get_current_user)]

@router.get("/me", response_model=UserPublic)
def read_user_me(current_user: CurrentUser) -> Any:
    return current_user  # 无需手动处理认证逻辑
  • Annotated 是 Python 3.9+ 引入的标准库类型注解工具，用于给类型附加元数据（如依赖、验证规则）
  • Depends FastAPI 的依赖注入核心：声明 这个参数的值需要调用指定函数来获取
  • CurrentUser 类型别名是对 User 类型和依赖注入的结合，其核心逻辑是：表示的类型是 User（用户模型），值通过 get_current_user() 函数获取
  • 后面再 read_user_me 函数中，就无需手动编写获取当前用户的逻辑，只需通过 current_user 参数即可获取到当前登录的用户信息，FastAPI 会自动调用 get_current_user 函数来填充这个参数

这种写法的核心优势

  • 代码复用性极高：不同接口无需重复编写 Token 解析、用户查询逻辑；
  • 业务逻辑与通用逻辑解耦：认证、数据库连接等通用逻辑封装在依赖函数中，业务代码只需关注核心功能
  • 测试时可以通过 Depends 传入模拟数据，方便单元测试

如下展示了两种写法的对比，一种是直接在函数中处理认证逻辑，另一种是通过依赖注入自动填充：

1
2
3
4
5
6
7
# ❌ 不推荐
@router.get("/me")
def read_user_me(token: str = Depends(oauth2_scheme)):
    user = verify_token(token)
    if not user:
        raise HTTPException(401)
    return user

该项目的做法：

1
2
3
4
# ✅ 推荐：类型化依赖
@router.get("/me")
def read_user_me(current_user: CurrentUser):
    return current_user

另外，这个项目实现了两种权限系统：超级用户和普通用户，同样可以通过依赖注入的方式实现不同接口的不同权限控制：

1
2
3
4
5
6
7
8
9
# 超级管理员权限
@router.get("/", dependencies=[Depends(get_current_active_superuser)])
def read_users(session: SessionDep) -> Any:
    ...

# 普通用户权限（自动注入 CurrentUser）
@router.get("/me")
def read_user_me(current_user: CurrentUser) -> Any:
    ...

jwt 认证

这个项目里后端逻辑比较复杂的流程就是用户认证，这里总结一下该项目是如何实现用户登录认证流程的。

  • 登录流程：验证用户密码后会返回 JWT Token，JWT Token 是一个加密字符串，包含了用户信息、过期时间等
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
POST /api/v1/login/access-token
        │
        ▼
┌─────────────────────────────────────┐
│ OAuth2PasswordRequestForm 解析       │
│ username=email, password=xxx        │
└─────────────────────────────────────┘
        │
        ▼
┌─────────────────────────────────────┐
│ crud.authenticate(session, email,   │
│                   password)         │
└─────────────────────────────────────┘
        │
        ├──── get_user_by_email() ──→ SELECT * FROM "user" WHERE email=?
        │
        ├──── 用户不存在 ──→ verify_password(password, DUMMY_HASH)
        │                      │
        │                      └── 防时序攻击，恒定时间返回
        │
        └──── 用户存在 ──→ verify_password(password, hashed_password)
                           │
                           ├── 验证失败 → return None
                           │
                           └── 验证成功 → 检查是否需要升级哈希
                                          │
                                          └── return User
        │
        ▼
┌─────────────────────────────────────┐
│ 用户验证通过                          │
│ - is_active 检查                    │
│ - 生成 JWT Token                    │
└─────────────────────────────────────┘
        │
        ▼
┌─────────────────────────────────────┐
│ 返回 Token 响应                      │
│ { "access_token": "eyJ...",         │
│   "token_type": "bearer" }          │
└─────────────────────────────────────┘
  • Token 验证流程：
1
2
3
4
┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│ API请求   │───→│ Token提取│───→| JWT解码  │───→ │ 用户查询  |
│ + Token  │    │          │    │          │    │          │
└──────────┘    └──────────┘    └──────────┘    └──────────┘
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
# backend/app/api/deps.py

# 1. OAuth2 Bearer Token 提取
reusable_oauth2 = OAuth2PasswordBearer(
    tokenUrl=f"{settings.API_V1_STR}/login/access-token"
)

# 2. 类型别名定义
CurrentUser = Annotated[User, Depends(get_current_user)]


# 3. 依赖注入：获取当前用户
def get_current_user(session: SessionDep, token: TokenDep) -> User:
    # Step 1: JWT 解码
    try:
        payload = jwt.decode(
            token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
        )
        token_data = TokenPayload(**payload)
    except (InvalidTokenError, ValidationError):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)

    # Step 2: 查询用户
    user = session.get(User, token_data.sub)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    # Step 3: 状态验证
    if not user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")

    return user

其他模块

crud.py 中封装了数据库的 CURD 操作，实现了对 User、Item 等业务的增删改查：

1
2
3
4
5
6
7
8
9
10
11
# 实现 crud 的数据库 repo 操作
def create_user(*, session: Session, user_create: UserCreate) -> User:
    db_obj = User.model_validate(
        user_create, update={"hashed_password": get_password_hash(user_create.password)}
    )
    session.add(db_obj)
    session.commit()
    session.refresh(db_obj)
    return db_obj

......

models.py 中定义了数据模型，同时包含了 API 请求/响应数据模型、数据库存储数据模型，因为 API 的数据结构可能和数据库存储的数据结构有很多重复字段，通过类继承的方式，可以减少重复代码。这其实也是使用 SQLModel 库的好处之一，减少 API 请求 schema 和数据库模型之间的重复代码。

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
# Shared properties
class ItemBase(SQLModel):
    title: str = Field(min_length=1, max_length=255)
    description: str | None = Field(default=None, max_length=255)


# Properties to receive on item creation
class ItemCreate(ItemBase):
    pass


# Properties to receive on item update
class ItemUpdate(ItemBase):
    title: str | None = Field(default=None, min_length=1, max_length=255)  # type: ignore


# Database model, database table inferred from class name
class Item(ItemBase, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    created_at: datetime | None = Field(
        default_factory=get_datetime_utc,
        sa_type=DateTime(timezone=True),  # type: ignore
    )
    owner_id: uuid.UUID = Field(
        foreign_key="user.id", nullable=False, ondelete="CASCADE"
    )
    owner: User | None = Relationship(back_populates="items")

数据库迁移

该项目使用 Alembic 作为数据库迁移工具，它是 SQLAlchemy 生态的官方迁移方案，与 SQLModel 无缝集成。

特性 说明
版本控制 每次迁移都有唯一 revision ID，形成链式依赖
自动生成 对比模型与数据库，自动生成迁移脚本
可回滚 每个迁移都支持 upgrade 和 downgrade
多环境支持 开发/测试/生产环境统一的迁移流程

迁移文件结构：

1
2
3
4
5
6
7
8
9
10
11
12
13
backend/
├── alembic.ini                    # Alembic 配置文件
└── app/
    ├── alembic/
    │   ├── env.py                 # 迁移环境配置
    │   ├── script.py.mako         # 迁移脚本模板
    │   └── versions/              # 迁移版本文件目录
    │       ├── e2412789c190_initialize_models.py
    │       ├── d98dd8ec85a3_edit_replace_id_integers_in_all_models_.py
    │       ├── 9c0a54914c78_add_max_length_for_string_varchar_.py
    │       ├── 1a31ce608336_add_cascade_delete_relationships.py
    │       └── fe56fa70289e_add_created_at_to_user_and_item.py
    └── models.py                  # 数据模型定义

它构成了如下迁移链：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
None (初始)
    │
    ▼
e2412789c190 ──► 初始化模型 (User, Item，整数ID)
    │
    ▼
9c0a54914c78 ──► 添加字符串字段长度限制
    │
    ▼
d98dd8ec85a3 ──► ID 从整数改为 UUID
    │
    ▼
1a31ce608336 ──► 添加级联删除关系
    │
    ▼
fe56fa70289e ──► 添加 created_at 字段
    │
    ▼
  head (最新版本)

在开发阶段，如果修改了数据库，可以通过如下命令自动生成迁移脚本：

1
alembic revision --autogenerate -m "add user phone field"

在实际部署阶段：在 docker compose.yml 文件中提供了一个 prestart 服务，专门用于执行迁移和数据库初始化。

1
2
3
4
5
6
7
8
9
10
11
12
13
prestart:
  image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}'
  build:
    context: .
    dockerfile: backend/Dockerfile
  networks:
    - traefik-public
    - default
  depends_on:
    db:
      condition: service_healthy
      restart: true
  command: bash scripts/prestart.sh

它执行 scripts/prestart.sh 脚本，其内容如下

1
2
3
4
5
6
7
8
9
10
11
12
#!/usr/bin/env bash
set -e
set -x

# 1. 等待数据库就绪：简单建立和数据库的连接，确认数据库服务启动正常
python app/backend_pre_start.py

# 2. 执行迁移
alembic upgrade head

# 3. 创建初始数据，包含初始的管理员用户
python app/initial_data.py

小结

文本详细总结了 full-stack-fastapi-template 项目的后端实现逻辑，重点学习了如何基于 FastAPI 构建自己的 Web 应用后端，尤其是通过依赖注入实现了通用逻辑（认证、数据库会话）与业务代码的彻底解耦，同时也学习了使用 FastAPI 后应该如何分层组织自己的代码结构，另外也通过这个示例项目学习了如何通过 JWT 实现用户登录认证、如何使用 Alembic 进行数据库迁移等后端应用开发中常见的需求。