FastAPI 分层架构深度解析：从 Controller 到 Service 与 CRUD 层

引言

在构建现代 Web 应用时，一个清晰、可维护的代码结构至关重要。FastAPI 以其高性能和易用性著称，但官方文档通常侧重于路由和 Pydantic 模型，对于如何组织大型项目的代码结构着墨不多。许多开发者，尤其是从 Flask 或 Django 迁移过来的，常常会困惑：为什么需要分层？Controller、Service、Repository（或 CRUD 层）这些层都是做什么的？它们之间如何协作？

本文将深入探讨 FastAPI 项目中常见的分层架构模式。我们将从一个简单的单文件示例开始，逐步揭示其局限性，然后引入分层思想，详细讲解每一层的职责、存在的理由，并通过一个模拟的“用户管理系统”项目来演示如何实践。最后，我们会探讨这种分层架构如何提升代码的可测试性、可维护性和团队协作效率。

1. 从“面条式”代码到分层架构的必然性

1.1 一个典型的“面条式”FastAPI 示例

许多 FastAPI 入门教程会展示类似下面的代码，将所有逻辑都写在路由函数里：

fromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelimportsqlite3 app=FastAPI()classUserCreate(BaseModel):username:stremail:str# 初始化数据库（简陋版）definit_db():conn=sqlite3.connect('test.db')c=conn.cursor()c.execute('''CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE, email TEXT)''')conn.commit()conn.close()init_db()@app.post("/users/")asyncdefcreate_user(user:UserCreate):"""创建用户"""conn=sqlite3.connect('test.db')c=conn.cursor()# 1. 业务逻辑：检查用户名是否已存在c.execute("SELECT id FROM users WHERE username = ?",(user.username,))ifc.fetchone():conn.close()raiseHTTPException(status_code=400,detail="Username already exists")# 2. 数据操作：插入新用户c.execute("INSERT INTO users (username, email) VALUES (?, ?)",(user.username,user.email))conn.commit()new_id=c.lastrowid conn.close()# 3. 响应格式化return{"id":new_id,"username":user.username,"email":user.email}@app.get("/users/{user_id}")asyncdefget_user(user_id:int):"""获取用户"""conn=sqlite3.connect('test.db')c=conn.cursor()c.execute("SELECT id, username, email FROM users WHERE id = ?",(user_id,))row=c.fetchone()conn.close()ifnotrow:raiseHTTPException(status_code=404,detail="User not found")# 直接返回数据库行return{"id":row[0],"username":row[1],"email":row[2]}

问题分析：

1. 职责混杂：一个函数同时处理 HTTP 请求、业务逻辑验证（用户名查重）、数据库操作（SQL 执行）和响应格式化。这违反了“单一职责原则”。
2. 难以测试：要测试“用户名重复”的业务逻辑，你必须启动一个完整的 FastAPI 应用并连接真实数据库，这是集成测试，而非单元测试。
3. 代码重复：每个路由函数都要写数据库连接、关闭的样板代码。
4. 难以维护：如果未来想把 SQLite 换成 PostgreSQL，或者修改用户表的字段，你需要修改每一个涉及数据库操作的路由函数。
5. 无法复用：业务逻辑（如“用户名查重”）被牢牢绑定在特定的 HTTP 端点里，无法在其他地方（如 CLI 命令、后台任务）使用。

1.2 分层的核心思想：关注点分离

分层架构的核心是“关注点分离”。我们将一个复杂的系统按照不同的“关注点”或“职责”拆分成多个层次，每一层只负责一件事，并通过清晰的接口与上下层通信。

对于典型的 Web 后端应用，我们可以抽象出以下几个核心关注点：

1. HTTP/Web 层：负责接收 HTTP 请求，解析参数，返回 HTTP 响应。这是与外部世界（客户端）的边界。
2. 业务逻辑层：这是应用的核心，包含所有的业务规则、流程和计算。例如，“创建用户前必须先检查邮箱格式和唯一性”。
3. 数据访问层：负责与持久化存储（数据库、缓存、外部 API）打交道，执行 CRUD（创建、读取、更新、删除）操作。

将这三者分离，就形成了最常见的三层架构。在 FastAPI 社区中，这三层通常被命名为：

• Controller / Router / API Layer：对应 HTTP/Web 层。
• Service Layer：对应业务逻辑层。
• Repository / CRUD / DAO Layer：对应数据访问层。

接下来，我们自底向上，详细剖析每一层。

2. 基石：CRUD/Repository 层（数据访问层）

2.1 职责与目标

这一层是唯一直接与数据库（或其他数据源）交互的地方。它的核心职责是：

• 封装所有数据持久化细节（SQL 语句、ORM 调用、连接池管理）。
• 提供一套简单的、面向对象的接口（如create,get,update,delete）给上层使用。
• 将数据库中的“行”转换为应用中的“模型对象”，反之亦然。

目标：让上层（Service）完全不用关心数据是如何存储和获取的。Service 层只需要说“给我保存这个用户”，而不用管是用 SQLite、MySQL 还是 MongoDB。

2.2 为什么需要它？

1. 集中变化点：当数据库 Schema 变更或更换数据库产品时，你只需要修改这一层的代码。
2. 避免 SQL 注入：通过使用参数化查询或 ORM，将安全风险隔离在这一层。
3. 便于测试：你可以为这一层创建一个“模拟”版本（Mock/Fake），在测试 Service 层时，无需真实的数据库。
4. 代码复用：多个不同的 Service 可以共用同一个 Repository。

2.3 实现示例：UserRepository

我们首先定义领域模型（通常是一个简单的 Python 类或 Pydantic 模型）：

# app/models/user.pyfrompydanticimportBaseModelfromtypingimportOptionalclassUserInDB(BaseModel):"""代表数据库中存储的用户模型"""id:Optional[int]=None# 数据库自增 IDusername:stremail:strclassConfig:orm_mode=True# 如果使用 SQLAlchemy 等 ORM 时需要

然后实现 Repository：

# app/repositories/user_repository.pyimportsqlite3fromtypingimportList,Optionalfromapp.models.userimportUserInDBclassUserRepository:"""用户数据仓库，负责所有与`users`表相关的数据库操作"""def__init__(self,db_path:str="test.db"):self.db_path=db_pathdef_get_connection(self):"""获取数据库连接（简单示例，生产环境应用连接池）"""returnsqlite3.connect(self.db_path)defcreate(self,user:UserInDB)->UserInDB:"""创建用户记录"""conn=self._get_connection()cursor=conn.cursor()try:cursor.execute("INSERT INTO users (username, email) VALUES (?, ?)",(user.username,user.email))conn.commit()user.id=cursor.lastrowidreturnuserfinally:conn.close()defget_by_id(self,user_id:int)->Optional[UserInDB]:"""根据ID获取用户"""conn=self._get_connection()cursor=conn.cursor()try:cursor.execute("SELECT id, username, email FROM users WHERE id = ?",(user_id,))row=cursor.fetchone()ifrow:returnUserInDB(id=row[0],username=row[1],email=row[2])returnNonefinally:conn.close()defget_by_username(self,username:str)->Optional[UserInDB]:"""根据用户名获取用户"""conn=self._get_connection()cursor=conn.cursor()try:cursor.execute("SELECT id, username, email FROM users WHERE username = ?",(username,))row=cursor.fetchone()ifrow:returnUserInDB(id=row[0],username=row[1],email=row[2])returnNonefinally:conn.close()# 还可以实现 update, delete, list_all 等方法

关键点：

• 所有方法都返回或操作UserInDB模型对象，而不是原始的元组或字典。
• 数据库连接的生命周期被封装在方法内部。
• 提供了基于业务含义的查询方法（如get_by_username），而不仅仅是通用的get。

3. 核心：Service 层（业务逻辑层）

3.1 职责与目标

Service 层是应用的“大脑”，它包含了所有的业务规则和流程。它的职责是：

• 协调多个 Repository 的操作来完成一个业务用例（如“用户注册”可能涉及用户表、验证码表）。
• 执行业务验证（如“邮箱格式是否正确”、“用户是否已存在”）。
• 处理业务计算和转换。
• 管理事务（如果需要确保多个数据库操作原子性）。

目标：封装复杂的业务逻辑，使其独立于 Web 框架和数据库。一个 Service 方法应该完整地代表一个“业务用例”。

3.2 为什么需要它？

1. 业务逻辑集中地：所有关于“如何创建用户”的规则都放在UserService.create_user里，而不是散落在各个 Controller 中。这避免了重复和矛盾。
2. 框架无关性：Service 层不导入fastapi或sqlite3。这意味着同一套业务逻辑可以轻松地被 CLI 工具、后台任务、GraphQL 接口甚至另一个 Web 框架（如 Django）复用。
3. 可测试性极强：因为 Service 层只依赖 Repository 接口（而非具体实现），我们可以轻松地传入 Mock Repository 来测试各种业务场景（如用户已存在、邮箱无效等），而无需启动 Web 服务器和数据库。
4. 清晰的用例边界：每个 Service 方法对应一个明确的用户或系统操作。

3.3 实现示例：UserService

# app/services/user_service.pyfromtypingimportOptionalfromapp.models.userimportUserInDBfromapp.repositories.user_repositoryimportUserRepositoryfrompydanticimportEmailStr,ValidationErrorclassUserService:"""用户业务服务，包含所有与用户相关的业务逻辑"""def__init__(self,user_repository:UserRepository):# 依赖注入：Service 依赖于一个抽象的 Repository 接口self.user_repo=user_repositorydefcreate_user(self,username:str,email:str)->UserInDB:""" 创建用户的完整业务逻辑 1. 验证输入 2. 检查唯一性约束 3. 调用 Repository 保存 4. 返回结果 """# 1. 业务验证（简单示例）ifnotusernameorlen(username)<3:raiseValueError("用户名至少需要3个字符")try:# 使用 Pydantic 验证邮箱格式_=EmailStr.validate(email)exceptValidationError:raiseValueError("邮箱格式无效")# 2. 业务规则：检查用户名是否已存在existing_user=self.user_repo.get_by_username(username)ifexisting_user:raiseValueError(f"用户名 '{username}' 已被占用")# 3. 创建领域对象new_user=UserInDB(username=username,email=email)# 4. 调用 Repository 持久化created_user=self.user_repo.create(new_user)# 5. 可选的后续业务操作（如发送欢迎邮件、记录日志等）# self._send_welcome_email(created_user.email)returncreated_userdefget_user(self,user_id:int)->Optional[UserInDB]:"""获取用户信息（简单的透传，但未来可在此添加权限检查等逻辑）"""returnself.user_repo.get_by_id(user_id)# 其他业务方法：update_user, delete_user, search_users 等

关键点：

• Service 通过构造函数接收一个UserRepository实例。这是一种依赖注入，使得测试时可以轻松替换为 Mock 对象。
• 所有业务错误都通过抛出 Python 原生异常（如ValueError）来表示，而不是 HTTP 异常。这保持了层的纯粹性。
• Service 方法包含了完整的业务用例流程。

4. 门面：Controller/Router 层（Web/HTTP 层）

4.1 职责与目标

Controller 层是系统对外的接口和适配器。它的职责是：

• 定义 HTTP 路由（URL 路径和方法）。
• 解析 HTTP 请求参数（路径参数、查询参数、请求体）。
• 调用合适的 Service 方法来处理请求。
• 将 Service 层的返回结果或异常，转换为合适的 HTTP 响应（状态码、JSON 体）。
• 处理跨领域关注点，如认证、授权、请求日志、限流等（通常通过 FastAPI 依赖项实现）。

目标：做一个“薄”的层，主要做协议转换和适配工作。它不应该包含任何业务逻辑。

4.2 为什么需要它？

1. 协议隔离：将 HTTP 协议的细节（如状态码、Header、Cookie）与核心业务逻辑隔离开。如果未来需要增加 GraphQL 或 gRPC 接口，可以复用 Service 层。
2. 框架专长：利用 FastAPI 强大的功能，如自动请求验证（Pydantic）、OpenAPI 文档生成、依赖注入系统。
3. 统一错误处理：在此层集中捕获 Service 层抛出的业务异常，并将其映射为对应的 HTTP 错误响应（如ValueError->400 Bad Request, 查找不到 ->404 Not Found）。

4.3 实现示例：UserController (FastAPI Router)

# app/api/v1/endpoints/users.pyfromfastapiimportAPIRouter,Depends,HTTPException,statusfrompydanticimportBaseModelfromapp.services.user_serviceimportUserServicefromapp.repositories.user_repositoryimportUserRepository# 定义请求/响应模型（属于API契约）classUserCreateRequest(BaseModel):username:stremail:strclassUserResponse(BaseModel):id:intusername:stremail:str# 创建路由router=APIRouter(prefix="/users",tags=["users"])# 依赖项：创建 Service 和 Repository 实例（生产环境会用更复杂的依赖注入容器）defget_user_service()->UserService:repo=UserRepository()returnUserService(repo)@router.post("/",response_model=UserResponse,status_code=status.HTTP_201_CREATED)asyncdefcreate_user(user_data:UserCreateRequest,user_service:UserService=Depends(get_user_service)):""" 创建新用户 """try:# 1. 调用 Service 层执行业务逻辑created_user=user_service.create_user(username=user_data.username,email=user_data.email)# 2. 将领域对象转换为 API 响应模型returnUserResponse(id=created_user.id,username=created_user.username,email=created_user.email)exceptValueErrorase:# 3. 捕获业务异常，转换为 HTTP 异常raiseHTTPException(status_code=status.HTTP_400_BAD_REQUEST,detail=str(e))# 其他异常（如数据库连接错误）可以由 FastAPI 的全局异常处理器处理@router.get("/{user_id}",response_model=UserResponse)asyncdefget_user(user_id:int,user_service:UserService=Depends(get_user_service)):""" 根据ID获取用户 """user=user_service.get_user(user_id)ifnotuser:raiseHTTPException(status_code=status.HTTP_404_NOT_FOUND,detail="User not found")returnUserResponse(id=user.id,username=user.username,email=user.email)

关键点：

• 每个端点函数都非常“薄”，主要工作是“转换”和“委托”。
• 使用 FastAPI 的Depends进行依赖注入，获取 Service 实例。
• 定义专门的请求/响应模型（UserCreateRequest,UserResponse），它们与内部的领域模型（UserInDB）是分离的。这提供了灵活性，例如 API 响应中可以省略某些字段或增加计算字段。

5. 项目结构全景与数据流

将以上各层组合起来，一个典型的分层 FastAPI 项目目录结构如下：

my_fastapi_project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用创建和路由注册入口 │ ├── core/ # 核心配置、依赖、工具 │ │ ├── config.py │ │ └── dependencies.py # 定义全局依赖项，如 get_db, get_user_service │ ├── models/ # Pydantic 领域模型/ORM 模型 │ │ └── user.py │ ├── schemas/ # API 请求/响应模型 (可选，可与 models 合并) │ │ └── user.py │ ├── repositories/ # 数据访问层 │ │ ├── __init__.py │ │ ├── base.py # 抽象基类 │ │ └── user_repository.py │ ├