引言在构建现代 Web 应用时一个清晰、可维护的代码结构至关重要。FastAPI 以其高性能和易用性著称但官方文档通常侧重于路由和 Pydantic 模型对于如何组织大型项目的代码结构着墨不多。许多开发者尤其是从 Flask 或 Django 迁移过来的常常会困惑为什么需要分层Controller、Service、Repository或 CRUD 层这些层都是做什么的它们之间如何协作本文将深入探讨 FastAPI 项目中常见的分层架构模式。我们将从一个简单的单文件示例开始逐步揭示其局限性然后引入分层思想详细讲解每一层的职责、存在的理由并通过一个模拟的“用户管理系统”项目来演示如何实践。最后我们会探讨这种分层架构如何提升代码的可测试性、可维护性和团队协作效率。1. 从“面条式”代码到分层架构的必然性1.1 一个典型的“面条式”FastAPI 示例许多 FastAPI 入门教程会展示类似下面的代码将所有逻辑都写在路由函数里fromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelimportsqlite3 appFastAPI()classUserCreate(BaseModel):username:stremail:str# 初始化数据库简陋版definit_db():connsqlite3.connect(test.db)cconn.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):创建用户connsqlite3.connect(test.db)cconn.cursor()# 1. 业务逻辑检查用户名是否已存在c.execute(SELECT id FROM users WHERE username ?,(user.username,))ifc.fetchone():conn.close()raiseHTTPException(status_code400,detailUsername already exists)# 2. 数据操作插入新用户c.execute(INSERT INTO users (username, email) VALUES (?, ?),(user.username,user.email))conn.commit()new_idc.lastrowid conn.close()# 3. 响应格式化return{id:new_id,username:user.username,email:user.email}app.get(/users/{user_id})asyncdefget_user(user_id:int):获取用户connsqlite3.connect(test.db)cconn.cursor()c.execute(SELECT id, username, email FROM users WHERE id ?,(user_id,))rowc.fetchone()conn.close()ifnotrow:raiseHTTPException(status_code404,detailUser not found)# 直接返回数据库行return{id:row[0],username:row[1],email:row[2]}问题分析职责混杂一个函数同时处理 HTTP 请求、业务逻辑验证用户名查重、数据库操作SQL 执行和响应格式化。这违反了“单一职责原则”。难以测试要测试“用户名重复”的业务逻辑你必须启动一个完整的 FastAPI 应用并连接真实数据库这是集成测试而非单元测试。代码重复每个路由函数都要写数据库连接、关闭的样板代码。难以维护如果未来想把 SQLite 换成 PostgreSQL或者修改用户表的字段你需要修改每一个涉及数据库操作的路由函数。无法复用业务逻辑如“用户名查重”被牢牢绑定在特定的 HTTP 端点里无法在其他地方如 CLI 命令、后台任务使用。1.2 分层的核心思想关注点分离分层架构的核心是“关注点分离”。我们将一个复杂的系统按照不同的“关注点”或“职责”拆分成多个层次每一层只负责一件事并通过清晰的接口与上下层通信。对于典型的 Web 后端应用我们可以抽象出以下几个核心关注点HTTP/Web 层负责接收 HTTP 请求解析参数返回 HTTP 响应。这是与外部世界客户端的边界。业务逻辑层这是应用的核心包含所有的业务规则、流程和计算。例如“创建用户前必须先检查邮箱格式和唯一性”。数据访问层负责与持久化存储数据库、缓存、外部 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 为什么需要它集中变化点当数据库 Schema 变更或更换数据库产品时你只需要修改这一层的代码。避免 SQL 注入通过使用参数化查询或 ORM将安全风险隔离在这一层。便于测试你可以为这一层创建一个“模拟”版本Mock/Fake在测试 Service 层时无需真实的数据库。代码复用多个不同的 Service 可以共用同一个 Repository。2.3 实现示例UserRepository我们首先定义领域模型通常是一个简单的 Python 类或 Pydantic 模型# app/models/user.pyfrompydanticimportBaseModelfromtypingimportOptionalclassUserInDB(BaseModel):代表数据库中存储的用户模型id:Optional[int]None# 数据库自增 IDusername:stremail:strclassConfig:orm_modeTrue# 如果使用 SQLAlchemy 等 ORM 时需要然后实现 Repository# app/repositories/user_repository.pyimportsqlite3fromtypingimportList,Optionalfromapp.models.userimportUserInDBclassUserRepository:用户数据仓库负责所有与users表相关的数据库操作def__init__(self,db_path:strtest.db):self.db_pathdb_pathdef_get_connection(self):获取数据库连接简单示例生产环境应用连接池returnsqlite3.connect(self.db_path)defcreate(self,user:UserInDB)-UserInDB:创建用户记录connself._get_connection()cursorconn.cursor()try:cursor.execute(INSERT INTO users (username, email) VALUES (?, ?),(user.username,user.email))conn.commit()user.idcursor.lastrowidreturnuserfinally:conn.close()defget_by_id(self,user_id:int)-Optional[UserInDB]:根据ID获取用户connself._get_connection()cursorconn.cursor()try:cursor.execute(SELECT id, username, email FROM users WHERE id ?,(user_id,))rowcursor.fetchone()ifrow:returnUserInDB(idrow[0],usernamerow[1],emailrow[2])returnNonefinally:conn.close()defget_by_username(self,username:str)-Optional[UserInDB]:根据用户名获取用户connself._get_connection()cursorconn.cursor()try:cursor.execute(SELECT id, username, email FROM users WHERE username ?,(username,))rowcursor.fetchone()ifrow:returnUserInDB(idrow[0],usernamerow[1],emailrow[2])returnNonefinally:conn.close()# 还可以实现 update, delete, list_all 等方法关键点所有方法都返回或操作UserInDB模型对象而不是原始的元组或字典。数据库连接的生命周期被封装在方法内部。提供了基于业务含义的查询方法如get_by_username而不仅仅是通用的get。3. 核心Service 层业务逻辑层3.1 职责与目标Service 层是应用的“大脑”它包含了所有的业务规则和流程。它的职责是协调多个 Repository 的操作来完成一个业务用例如“用户注册”可能涉及用户表、验证码表。执行业务验证如“邮箱格式是否正确”、“用户是否已存在”。处理业务计算和转换。管理事务如果需要确保多个数据库操作原子性。目标封装复杂的业务逻辑使其独立于 Web 框架和数据库。一个 Service 方法应该完整地代表一个“业务用例”。3.2 为什么需要它业务逻辑集中地所有关于“如何创建用户”的规则都放在UserService.create_user里而不是散落在各个 Controller 中。这避免了重复和矛盾。框架无关性Service 层不导入fastapi或sqlite3。这意味着同一套业务逻辑可以轻松地被 CLI 工具、后台任务、GraphQL 接口甚至另一个 Web 框架如 Django复用。可测试性极强因为 Service 层只依赖 Repository 接口而非具体实现我们可以轻松地传入 Mock Repository 来测试各种业务场景如用户已存在、邮箱无效等而无需启动 Web 服务器和数据库。清晰的用例边界每个 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_repouser_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_userself.user_repo.get_by_username(username)ifexisting_user:raiseValueError(f用户名 {username} 已被占用)# 3. 创建领域对象new_userUserInDB(usernameusername,emailemail)# 4. 调用 Repository 持久化created_userself.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 为什么需要它协议隔离将 HTTP 协议的细节如状态码、Header、Cookie与核心业务逻辑隔离开。如果未来需要增加 GraphQL 或 gRPC 接口可以复用 Service 层。框架专长利用 FastAPI 强大的功能如自动请求验证Pydantic、OpenAPI 文档生成、依赖注入系统。统一错误处理在此层集中捕获 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# 创建路由routerAPIRouter(prefix/users,tags[users])# 依赖项创建 Service 和 Repository 实例生产环境会用更复杂的依赖注入容器defget_user_service()-UserService:repoUserRepository()returnUserService(repo)router.post(/,response_modelUserResponse,status_codestatus.HTTP_201_CREATED)asyncdefcreate_user(user_data:UserCreateRequest,user_service:UserServiceDepends(get_user_service)): 创建新用户 try:# 1. 调用 Service 层执行业务逻辑created_useruser_service.create_user(usernameuser_data.username,emailuser_data.email)# 2. 将领域对象转换为 API 响应模型returnUserResponse(idcreated_user.id,usernamecreated_user.username,emailcreated_user.email)exceptValueErrorase:# 3. 捕获业务异常转换为 HTTP 异常raiseHTTPException(status_codestatus.HTTP_400_BAD_REQUEST,detailstr(e))# 其他异常如数据库连接错误可以由 FastAPI 的全局异常处理器处理router.get(/{user_id},response_modelUserResponse)asyncdefget_user(user_id:int,user_service:UserServiceDepends(get_user_service)): 根据ID获取用户 useruser_service.get_user(user_id)ifnotuser:raiseHTTPException(status_codestatus.HTTP_404_NOT_FOUND,detailUser not found)returnUserResponse(iduser.id,usernameuser.username,emailuser.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 │ ├
FastAPI 分层架构深度解析：从 Controller 到 Service 与 CRUD 层
引言在构建现代 Web 应用时一个清晰、可维护的代码结构至关重要。FastAPI 以其高性能和易用性著称但官方文档通常侧重于路由和 Pydantic 模型对于如何组织大型项目的代码结构着墨不多。许多开发者尤其是从 Flask 或 Django 迁移过来的常常会困惑为什么需要分层Controller、Service、Repository或 CRUD 层这些层都是做什么的它们之间如何协作本文将深入探讨 FastAPI 项目中常见的分层架构模式。我们将从一个简单的单文件示例开始逐步揭示其局限性然后引入分层思想详细讲解每一层的职责、存在的理由并通过一个模拟的“用户管理系统”项目来演示如何实践。最后我们会探讨这种分层架构如何提升代码的可测试性、可维护性和团队协作效率。1. 从“面条式”代码到分层架构的必然性1.1 一个典型的“面条式”FastAPI 示例许多 FastAPI 入门教程会展示类似下面的代码将所有逻辑都写在路由函数里fromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelimportsqlite3 appFastAPI()classUserCreate(BaseModel):username:stremail:str# 初始化数据库简陋版definit_db():connsqlite3.connect(test.db)cconn.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):创建用户connsqlite3.connect(test.db)cconn.cursor()# 1. 业务逻辑检查用户名是否已存在c.execute(SELECT id FROM users WHERE username ?,(user.username,))ifc.fetchone():conn.close()raiseHTTPException(status_code400,detailUsername already exists)# 2. 数据操作插入新用户c.execute(INSERT INTO users (username, email) VALUES (?, ?),(user.username,user.email))conn.commit()new_idc.lastrowid conn.close()# 3. 响应格式化return{id:new_id,username:user.username,email:user.email}app.get(/users/{user_id})asyncdefget_user(user_id:int):获取用户connsqlite3.connect(test.db)cconn.cursor()c.execute(SELECT id, username, email FROM users WHERE id ?,(user_id,))rowc.fetchone()conn.close()ifnotrow:raiseHTTPException(status_code404,detailUser not found)# 直接返回数据库行return{id:row[0],username:row[1],email:row[2]}问题分析职责混杂一个函数同时处理 HTTP 请求、业务逻辑验证用户名查重、数据库操作SQL 执行和响应格式化。这违反了“单一职责原则”。难以测试要测试“用户名重复”的业务逻辑你必须启动一个完整的 FastAPI 应用并连接真实数据库这是集成测试而非单元测试。代码重复每个路由函数都要写数据库连接、关闭的样板代码。难以维护如果未来想把 SQLite 换成 PostgreSQL或者修改用户表的字段你需要修改每一个涉及数据库操作的路由函数。无法复用业务逻辑如“用户名查重”被牢牢绑定在特定的 HTTP 端点里无法在其他地方如 CLI 命令、后台任务使用。1.2 分层的核心思想关注点分离分层架构的核心是“关注点分离”。我们将一个复杂的系统按照不同的“关注点”或“职责”拆分成多个层次每一层只负责一件事并通过清晰的接口与上下层通信。对于典型的 Web 后端应用我们可以抽象出以下几个核心关注点HTTP/Web 层负责接收 HTTP 请求解析参数返回 HTTP 响应。这是与外部世界客户端的边界。业务逻辑层这是应用的核心包含所有的业务规则、流程和计算。例如“创建用户前必须先检查邮箱格式和唯一性”。数据访问层负责与持久化存储数据库、缓存、外部 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 为什么需要它集中变化点当数据库 Schema 变更或更换数据库产品时你只需要修改这一层的代码。避免 SQL 注入通过使用参数化查询或 ORM将安全风险隔离在这一层。便于测试你可以为这一层创建一个“模拟”版本Mock/Fake在测试 Service 层时无需真实的数据库。代码复用多个不同的 Service 可以共用同一个 Repository。2.3 实现示例UserRepository我们首先定义领域模型通常是一个简单的 Python 类或 Pydantic 模型# app/models/user.pyfrompydanticimportBaseModelfromtypingimportOptionalclassUserInDB(BaseModel):代表数据库中存储的用户模型id:Optional[int]None# 数据库自增 IDusername:stremail:strclassConfig:orm_modeTrue# 如果使用 SQLAlchemy 等 ORM 时需要然后实现 Repository# app/repositories/user_repository.pyimportsqlite3fromtypingimportList,Optionalfromapp.models.userimportUserInDBclassUserRepository:用户数据仓库负责所有与users表相关的数据库操作def__init__(self,db_path:strtest.db):self.db_pathdb_pathdef_get_connection(self):获取数据库连接简单示例生产环境应用连接池returnsqlite3.connect(self.db_path)defcreate(self,user:UserInDB)-UserInDB:创建用户记录connself._get_connection()cursorconn.cursor()try:cursor.execute(INSERT INTO users (username, email) VALUES (?, ?),(user.username,user.email))conn.commit()user.idcursor.lastrowidreturnuserfinally:conn.close()defget_by_id(self,user_id:int)-Optional[UserInDB]:根据ID获取用户connself._get_connection()cursorconn.cursor()try:cursor.execute(SELECT id, username, email FROM users WHERE id ?,(user_id,))rowcursor.fetchone()ifrow:returnUserInDB(idrow[0],usernamerow[1],emailrow[2])returnNonefinally:conn.close()defget_by_username(self,username:str)-Optional[UserInDB]:根据用户名获取用户connself._get_connection()cursorconn.cursor()try:cursor.execute(SELECT id, username, email FROM users WHERE username ?,(username,))rowcursor.fetchone()ifrow:returnUserInDB(idrow[0],usernamerow[1],emailrow[2])returnNonefinally:conn.close()# 还可以实现 update, delete, list_all 等方法关键点所有方法都返回或操作UserInDB模型对象而不是原始的元组或字典。数据库连接的生命周期被封装在方法内部。提供了基于业务含义的查询方法如get_by_username而不仅仅是通用的get。3. 核心Service 层业务逻辑层3.1 职责与目标Service 层是应用的“大脑”它包含了所有的业务规则和流程。它的职责是协调多个 Repository 的操作来完成一个业务用例如“用户注册”可能涉及用户表、验证码表。执行业务验证如“邮箱格式是否正确”、“用户是否已存在”。处理业务计算和转换。管理事务如果需要确保多个数据库操作原子性。目标封装复杂的业务逻辑使其独立于 Web 框架和数据库。一个 Service 方法应该完整地代表一个“业务用例”。3.2 为什么需要它业务逻辑集中地所有关于“如何创建用户”的规则都放在UserService.create_user里而不是散落在各个 Controller 中。这避免了重复和矛盾。框架无关性Service 层不导入fastapi或sqlite3。这意味着同一套业务逻辑可以轻松地被 CLI 工具、后台任务、GraphQL 接口甚至另一个 Web 框架如 Django复用。可测试性极强因为 Service 层只依赖 Repository 接口而非具体实现我们可以轻松地传入 Mock Repository 来测试各种业务场景如用户已存在、邮箱无效等而无需启动 Web 服务器和数据库。清晰的用例边界每个 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_repouser_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_userself.user_repo.get_by_username(username)ifexisting_user:raiseValueError(f用户名 {username} 已被占用)# 3. 创建领域对象new_userUserInDB(usernameusername,emailemail)# 4. 调用 Repository 持久化created_userself.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 为什么需要它协议隔离将 HTTP 协议的细节如状态码、Header、Cookie与核心业务逻辑隔离开。如果未来需要增加 GraphQL 或 gRPC 接口可以复用 Service 层。框架专长利用 FastAPI 强大的功能如自动请求验证Pydantic、OpenAPI 文档生成、依赖注入系统。统一错误处理在此层集中捕获 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# 创建路由routerAPIRouter(prefix/users,tags[users])# 依赖项创建 Service 和 Repository 实例生产环境会用更复杂的依赖注入容器defget_user_service()-UserService:repoUserRepository()returnUserService(repo)router.post(/,response_modelUserResponse,status_codestatus.HTTP_201_CREATED)asyncdefcreate_user(user_data:UserCreateRequest,user_service:UserServiceDepends(get_user_service)): 创建新用户 try:# 1. 调用 Service 层执行业务逻辑created_useruser_service.create_user(usernameuser_data.username,emailuser_data.email)# 2. 将领域对象转换为 API 响应模型returnUserResponse(idcreated_user.id,usernamecreated_user.username,emailcreated_user.email)exceptValueErrorase:# 3. 捕获业务异常转换为 HTTP 异常raiseHTTPException(status_codestatus.HTTP_400_BAD_REQUEST,detailstr(e))# 其他异常如数据库连接错误可以由 FastAPI 的全局异常处理器处理router.get(/{user_id},response_modelUserResponse)asyncdefget_user(user_id:int,user_service:UserServiceDepends(get_user_service)): 根据ID获取用户 useruser_service.get_user(user_id)ifnotuser:raiseHTTPException(status_codestatus.HTTP_404_NOT_FOUND,detailUser not found)returnUserResponse(iduser.id,usernameuser.username,emailuser.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 │ ├
