FastAPI整洁架构实战：分层设计与依赖注入构建可维护后端

1. 项目概述为什么我们需要一个“干净”的FastAPI后端架构如果你和我一样用FastAPI开发过几个项目从简单的API服务到稍具规模的后台系统大概率会经历这样一个过程一开始main.py里写几个路由函数直接调用数据库操作代码跑得飞快成就感满满。但随着功能模块增加你会发现路由函数越来越臃肿业务逻辑和数据库查询、外部服务调用、数据验证全都搅在一起。想改个数据库字段得把整个路由逻辑翻个底朝天。想加个缓存又得在业务代码里到处打补丁。单元测试更是难以下手因为所有东西都紧密耦合在一起。这时候你需要的不是一个新框架而是一个清晰、可维护的代码组织方式——这就是“整洁架构”Clean Architecture要解决的问题。最近在GitHub上看到一个名为Flaiers/fastapi-clean-architecture的项目它提供了一个基于FastAPI实现整洁架构的实战模板。这个项目不是又一个教你用app.get装饰器的教程而是直指后端工程化的核心如何构建一个边界清晰、依赖关系明确、易于测试和扩展的应用结构。对于从“玩具项目”迈向“生产级应用”的开发者来说理解并实践这套架构思想其价值远超掌握某个具体的库或语法糖。简单来说这个模板项目为我们演示了如何将FastAPI这个高性能的现代Web框架与Robert C. Martin提出的整洁架构理念相结合。它旨在解决我们日常开发中最头疼的问题代码随着需求增长而腐化。通过强制性地分离关注点它让我们的核心业务逻辑独立于Web框架、数据库和外部服务从而获得极高的可测试性和可维护性。接下来我将带你深入拆解这个项目的设计思路、核心实现并分享如何在实际项目中应用和调整这套架构避开我踩过的一些坑。2. 架构核心深入理解整洁架构的分层与依赖规则2.1 整洁架构的核心思想依赖倒置与边界隔离在深入代码之前我们必须先吃透整洁架构的“道”而非仅仅模仿其“术”。整洁架构或称“洋葱架构”其核心目标是将软件划分为不同的同心圆层次让业务逻辑居于核心并独立于外部细节。这些层次由内到外通常是实体Entities - 用例Use Cases - 接口适配器Interface Adapters - 框架和驱动Frameworks Drivers。最关键的规则是依赖关系规则源代码的依赖方向必须指向同心圆的内层即内层圆对外层圆一无所知。具体来说实体层最内层包含企业级业务规则和核心数据对象。它应该是纯Python对象POJO不依赖任何框架、数据库或外部服务。用例层包含应用特定的业务规则协调数据流向实体层或从实体层流出。它定义了系统能做什么如“创建用户”、“处理订单”。接口适配器层这一层将用例和实体转换成外部机构如数据库、Web框架方便使用的格式。例如这里会有Repository接口的具体实现如用SQLAlchemy操作PostgreSQL以及将内部实体转换为FastAPI响应模型的Pydantic Schema。框架和驱动层最外层包含所有具体的工具和框架如FastAPI本身、数据库驱动、缓存客户端、消息队列等。Flaiers/fastapi-clean-architecture项目正是基于这一思想构建的。它通过清晰的目录结构强制你遵守这些分层。当你尝试在“实体”中导入SQLAlchemy的Column时你的代码审查工具或你的团队规范就应该亮起红灯。2.2 项目结构解析从目录看架构约束让我们看看一个典型的基于此模板的项目结构可能是什么样的在原项目基础上进行通用化阐述src/ ├── core/ # 核心配置、常量、依赖注入容器 ├── domain/ # 领域层实体层 │ ├── entities/ # 业务实体纯Python类 │ └── repositories/ # 仓储接口抽象基类ABC ├── application/ # 应用层用例层 │ ├── use_cases/ # 用例/交互器 │ └── dto/ # 应用层数据传输对象 ├── infrastructure/ # 基础设施层接口适配器框架驱动层 │ ├── database/ # 数据库相关模型、迁移、具体仓储实现 │ ├── api/ # Web API相关路由、控制器、序列化 │ └── external/ # 外部服务客户端邮件、短信、第三方API └── main.py # 应用入口组装所有部件为什么这样设计domain/的纯粹性这里的entities是简单的dataclass或PydanticBaseModel只定义属性及其业务验证方法。repositories里是抽象类如class IUserRepository(ABC)只定义接口def save(user: User) - User:不涉及任何具体数据库操作。这保证了业务核心的稳定性无论底层从MySQL换到MongoDB还是从REST换到GraphQL领域层代码都无需改动。application/的协调作用用例类如CreateUserUseCase接收输入调用领域实体的方法执行业务规则并通过仓储接口持久化数据。它不知道数据存在哪里也不知道请求来自HTTP还是CLI命令。infrastructure/的实现职责这里是所有“脏活累活”的地方。database/models.py中定义SQLAlchemy的Table类database/repositories.py中实现具体的SqlAlchemyUserRepository。api/目录下controllers处理HTTP请求调用用例并将结果序列化为Pydantic模型返回。依赖方向在这里被反转基础设施层依赖并实现内层定义的抽象接口。实操心得理解“依赖注入”是关键这套架构能跑起来灵魂在于依赖注入DI。在main.py或core/dependencies.py中你会看到一个“组装”过程将SqlAlchemyUserRepository的实例注入到CreateUserUseCase中而CreateUserUseCase只声明它需要IUserRepository。这样在单元测试时你可以轻松注入一个模拟的Mock仓储从而独立测试用例的业务逻辑无需启动真实的数据库。这是实现“可测试性”的核心手段。3. 核心模块实现与实操要点3.1 领域实体与仓储模式定义稳定的业务核心我们以一个简单的“用户”领域为例。在src/domain/entities/user.py中你可能会这样定义from pydantic import BaseModel, EmailStr, field_validator from typing import Optional from datetime import datetime class User(BaseModel): id: Optional[int] None username: str email: EmailStr hashed_password: str is_active: bool True created_at: Optional[datetime] None field_validator(username) classmethod def validate_username(cls, v: str) - str: if len(v) 3: raise ValueError(用户名至少3个字符) if not v.isalnum(): raise ValueError(用户名只能包含字母和数字) return v def activate(self) - None: 业务规则激活用户 self.is_active True def change_password(self, new_hashed_password: str) - None: 业务规则修改密码 self.hashed_password new_hashed_password # 这里可以添加更复杂的规则如密码历史记录注意这里用的是Pydantic的BaseModel它提供了强大的数据验证和序列化能力非常适合做领域实体。实体包含数据和与之相关的最基本的业务规则验证、状态变更。接下来在src/domain/repositories/user_repository.py中定义抽象仓储from abc import ABC, abstractmethod from typing import List, Optional from ..entities.user import User class IUserRepository(ABC): 用户仓储接口定义数据持久化契约 abstractmethod async def get_by_id(self, user_id: int) - Optional[User]: pass abstractmethod async def get_by_username(self, username: str) - Optional[User]: pass abstractmethod async def save(self, user: User) - User: pass abstractmethod async def delete(self, user_id: int) - bool: pass这个接口只定义了“做什么”完全没提“怎么做”。它属于领域层因为它是业务逻辑的一部分业务需要存取用户。注意事项实体与数据库模型的区别很多初学者会混淆领域实体和数据库ORM模型如SQLAlchemy的DeclarativeBase。记住实体是业务概念ORM模型是技术实现细节。实体应该对数据库一无所知。在整洁架构中我们会在基础设施层创建一个UserModel类它负责映射到数据库表并在仓储实现中负责在User实体和UserModel对象之间进行转换。这种转换虽然会写一些“胶水代码”但换来了领域核心的纯粹和独立。3.2 应用层用例编排业务逻辑的指挥官用例是应用层的核心它代表一个具体的用户交互或系统操作。在src/application/use_cases/create_user.py中from typing import Dict, Any from ...domain.entities.user import User from ...domain.repositories.user_repository import IUserRepository from ..dto.create_user_input import CreateUserInput class CreateUserUseCase: 创建用户用例 def __init__(self, user_repo: IUserRepository): # 依赖注入仓储接口而非具体实现 self.user_repo user_repo async def execute(self, input_dto: CreateUserInput) - User: 执行创建用户的业务逻辑 1. 验证输入通常由Pydantic在DTO层完成 2. 检查业务规则如用户名是否唯一 3. 创建领域实体 4. 调用仓储保存 5. 返回结果 # 检查用户名是否已存在业务规则 existing_user await self.user_repo.get_by_username(input_dto.username) if existing_user: raise ValueError(f用户名 {input_dto.username} 已存在) # 创建领域实体这里可以进行密码哈希等操作也可作为实体的初始化逻辑 # 注意密码哈希通常被视为基础设施细节可以在用例中调用一个服务或作为实体创建的一部分 hashed_password self._hash_password(input_dto.password) new_user User( usernameinput_dto.username, emailinput_dto.email, hashed_passwordhashed_password, is_activeTrue ) # 调用仓储接口保存不关心是存到MySQL还是Redis saved_user await self.user_repo.save(new_user) return saved_user def _hash_password(self, raw_password: str) - str: # 这是一个基础设施细节的示例。更佳实践是将其抽象为一个密码服务接口在基础设施层实现。 # 这里为了简化直接模拟。 import hashlib return hashlib.sha256(raw_password.encode()).hexdigest()用例类CreateUserUseCase的职责非常单一。它接收一个输入DTOData Transfer Object执行业务规则如唯一性检查操作领域实体并通过抽象的仓储接口进行持久化。它不包含任何HTTP状态码、JSON序列化或SQL语句。应用层DTO(create_user_input.py) 是进入应用层的数据契约通常也很简单from pydantic import BaseModel, EmailStr class CreateUserInput(BaseModel): username: str email: EmailStr password: str3.3 基础设施层实现连接抽象与具体的桥梁这是代码量可能最大的层因为它包含了所有外部依赖的具体实现。首先实现具体的仓储(src/infrastructure/database/repositories/user_repository_impl.py)from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from ...domain.entities.user import User from ...domain.repositories.user_repository import IUserRepository from ..models.user_model import UserModel # SQLAlchemy模型 class SqlAlchemyUserRepository(IUserRepository): IUserRepository的SQLAlchemy实现 def __init__(self, session: AsyncSession): self.session session async def get_by_id(self, user_id: int) - User | None: result await self.session.execute( select(UserModel).where(UserModel.id user_id) ) user_model result.scalar_one_or_none() if not user_model: return None return self._to_entity(user_model) async def get_by_username(self, username: str) - User | None: # ... 类似get_by_id的实现 async def save(self, user: User) - User: # 判断是新增还是更新 if user.id is None: user_model UserModel(**user.dict(exclude{id})) self.session.add(user_model) await self.session.flush() # 获取生成的ID await self.session.refresh(user_model) user.id user_model.id else: # 更新逻辑先查询出模型再更新 pass return user def _to_entity(self, model: UserModel) - User: 将数据库模型转换为领域实体 return User( idmodel.id, usernamemodel.username, emailmodel.email, hashed_passwordmodel.hashed_password, is_activemodel.is_active, created_atmodel.created_at ) def _to_model(self, entity: User) - UserModel: 将领域实体转换为数据库模型用于保存 return UserModel( identity.id, usernameentity.username, emailentity.email, hashed_passwordentity.hashed_password, is_activeentity.is_active, created_atentity.created_at )然后实现Web控制器或称为处理器(src/infrastructure/api/controllers/user_controller.py)from fastapi import APIRouter, Depends, HTTPException, status from ...application.use_cases.create_user import CreateUserUseCase from ...application.dto.create_user_input import CreateUserInput from ....core.dependencies import get_user_repository, get_db_session from sqlalchemy.ext.asyncio import AsyncSession router APIRouter(prefix/users, tags[users]) router.post(/, status_codestatus.HTTP_201_CREATED) async def create_user( user_input: CreateUserInput, session: AsyncSession Depends(get_db_session) ): 创建用户端点。 1. 依赖注入数据库会话。 2. 根据会话创建具体仓储实例。 3. 实例化用例并执行。 4. 处理用例抛出的业务异常转换为HTTP异常。 5. 将返回的领域实体序列化为响应模型。 # 组装具体仓储 user_repo get_user_repository(session) # 实例化用例注入具体仓储 use_case CreateUserUseCase(user_repouser_repo) try: created_user await use_case.execute(user_input) except ValueError as e: # 捕获用例中的业务逻辑异常如用户名重复 raise HTTPException( status_codestatus.HTTP_400_BAD_REQUEST, detailstr(e) ) # 将领域实体转换为API响应模型可定义在api/schemas/user_schema.py中 # 这里简单返回实际项目中会用一个Pydantic模型过滤掉敏感字段如hashed_password return { id: created_user.id, username: created_user.username, email: created_user.email, is_active: created_user.is_active }控制器是基础设施层的一部分它“适配”外部的HTTP世界到内部的应用层。它负责解析HTTP请求FastAPI自动完成。依赖注入具体的资源如数据库会话。组装具体的仓储和用例。调用用例。捕获应用层抛出的业务异常并将其转换为适当的HTTP响应。将内部实体转换为对外的API响应格式。4. 依赖注入与应用组装让架构运转起来4.1 构建依赖注入容器整洁架构的威力很大程度上依赖于依赖注入DI来实现控制反转。我们不需要一个重量级的DI框架利用FastAPI的Depends和Python的简单设计就能实现。在src/core/dependencies.py中from typing import Annotated from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession from ..infrastructure.database.session import AsyncSessionLocal from ..domain.repositories.user_repository import IUserRepository from ..infrastructure.database.repositories.user_repository_impl import SqlAlchemyUserRepository async def get_db_session() - AsyncSession: 获取数据库会话的依赖项。 使用FastAPI的依赖注入系统为每个请求提供独立的会话。 async with AsyncSessionLocal() as session: try: yield session await session.commit() # 请求成功提交事务 except Exception: await session.rollback() # 发生异常回滚事务 raise finally: await session.close() def get_user_repository( session: Annotated[AsyncSession, Depends(get_db_session)] ) - IUserRepository: 获取用户仓储实例的依赖项。 它依赖于数据库会话并返回一个实现了IUserRepository接口的具体实例。 return SqlAlchemyUserRepository(sessionsession) # 可以继续定义其他仓储或服务的依赖项 # def get_email_service() - IEmailService: ...这里的关键是get_user_repository函数。它声明要获得一个IUserRepository你需要先提供一个AsyncSession。当FastAPI处理请求时它会自动解析这个依赖链最终将SqlAlchemyUserRepository的实例提供给控制器。在测试时我们可以轻松地提供一个模拟的会话和仓储。4.2 应用入口与路由注册最后在src/main.py中我们将所有部件组装起来from fastapi import FastAPI from .infrastructure.api.controllers import user_controller, product_controller from .core.config import settings from .infrastructure.database.session import create_tables app FastAPI(titlesettings.PROJECT_NAME, versionsettings.VERSION) # 注册路由 app.include_router(user_controller.router) app.include_router(product_controller.router) app.on_event(startup) async def on_startup(): 应用启动事件。 可以在这里创建数据库表仅用于演示生产环境应用迁移工具如Alembic。 await create_tables() print(Application started and tables are ready.) app.get(/health) async def health_check(): 健康检查端点 return {status: healthy} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)这个文件非常干净它的职责就是创建FastAPI应用实例注册所有路由来自基础设施层并配置一些生命周期事件。所有的业务逻辑都已经被妥善地组织在了更内层的领域层和应用层中。5. 测试策略分层测试保障代码质量整洁架构带来的一个巨大优势是可测试性。我们可以对各层进行独立的、隔离的测试。5.1 领域层测试纯业务逻辑的单元测试领域实体和业务规则的测试最简单因为它们不依赖任何外部资源。# tests/domain/entities/test_user.py import pytest from src.domain.entities.user import User from pydantic import ValidationError def test_user_entity_creation(): 测试用户实体创建与基础验证 user User(usernamealice123, emailaliceexample.com, hashed_passwordhash) assert user.username alice123 assert user.is_active is True # 测试默认值 def test_user_username_validation(): 测试用户名字段的自定义验证器 # 有效用户名 User(usernamealice123, emailab.com, hashed_passwordhash) # 用户名过短 with pytest.raises(ValueError, match用户名至少3个字符): User(usernameab, emailab.com, hashed_passwordhash) # 用户名包含非法字符 with pytest.raises(ValueError, match用户名只能包含字母和数字): User(usernamealice-123, emailab.com, hashed_passwordhash) def test_user_business_methods(): 测试实体的业务方法 user User(usernamealice, emailab.com, hashed_passwordold_hash, is_activeFalse) user.activate() assert user.is_active is True user.change_password(new_hash) assert user.hashed_password new_hash5.2 应用层测试模拟外部依赖用例测试需要模拟Mock仓储接口。使用pytest和unittest.mock可以轻松完成。# tests/application/use_cases/test_create_user.py import pytest from unittest.mock import AsyncMock, create_autospec from src.application.use_cases.create_user import CreateUserUseCase from src.application.dto.create_user_input import CreateUserInput from src.domain.entities.user import User from src.domain.repositories.user_repository import IUserRepository pytest.mark.asyncio async def test_create_user_success(): 测试成功创建用户的用例 # 1. 创建模拟仓储 mock_repo create_autospec(IUserRepository) # 模拟仓储的get_by_username返回None表示用户名可用 mock_repo.get_by_username.return_value None # 模拟仓储的save方法返回一个模拟的用户实体 saved_user User(id1, usernamealice, emailab.com, hashed_passwordhashed_pw) mock_repo.save.return_value saved_user # 2. 实例化用例注入模拟仓储 use_case CreateUserUseCase(user_repomock_repo) input_dto CreateUserInput(usernamealice, emailab.com, passwordsecret123) # 3. 执行用例 result await use_case.execute(input_dto) # 4. 断言 assert result.id 1 assert result.username alice # 验证仓储方法被以正确的参数调用 mock_repo.get_by_username.assert_called_once_with(alice) mock_repo.save.assert_called_once() # 检查传递给save的实体参数 saved_entity mock_repo.save.call_args[0][0] assert saved_entity.username alice assert saved_entity.hashed_password is not None # 密码应被哈希 pytest.mark.asyncio async def test_create_user_duplicate_username(): 测试创建用户时用户名重复的业务异常 mock_repo create_autospec(IUserRepository) # 模拟用户名已存在 existing_user User(id99, usernamealice, emailexistingb.com, hashed_passwordhash) mock_repo.get_by_username.return_value existing_user use_case CreateUserUseCase(user_repomock_repo) input_dto CreateUserInput(usernamealice, emailnewb.com, passwordsecret123) with pytest.raises(ValueError, match用户名 alice 已存在): await use_case.execute(input_dto) # 确保save方法没有被调用 mock_repo.save.assert_not_called()这种测试完全隔离了数据库运行速度极快且只关注业务逻辑是否正确。5.3 集成测试与API测试对于基础设施层如控制器和具体仓储我们需要进行集成测试。这通常会启动一个测试数据库或者使用内存数据库如SQLite。# tests/infrastructure/api/test_user_controller.py import pytest from fastapi.testclient import TestClient from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker from src.main import app # 导入FastAPI应用 from src.infrastructure.database.models import Base # 配置测试数据库例如内存SQLite TEST_DATABASE_URL sqliteaiosqlite:///:memory: test_engine create_async_engine(TEST_DATABASE_URL, echoFalse) TestingSessionLocal sessionmaker(test_engine, class_AsyncSession, expire_on_commitFalse) pytest.fixture async def test_db(): 为每个测试用例提供干净的数据库会话和表 async with test_engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async with TestingSessionLocal() as session: yield session async with test_engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) pytest.fixture def test_client(test_db): 重写应用的依赖项使其使用测试数据库 # 这里需要临时替换main.py中get_db_session的依赖返回值为test_db # 一种常见做法是在app.dependency_overrides中覆盖 async def override_get_db(): yield test_db app.dependency_overrides[get_db_session] override_get_db with TestClient(app) as client: yield client app.dependency_overrides.clear() pytest.mark.asyncio async def test_create_user_endpoint(test_client): 测试创建用户的API端点 user_data { username: testuser, email: testexample.com, password: strongpassword } response test_client.post(/users/, jsonuser_data) assert response.status_code 201 data response.json() assert data[username] user_data[username] assert hashed_password not in data # 敏感信息不应暴露 assert data[is_active] is True # 测试重复用户名 response_duplicate test_client.post(/users/, jsonuser_data) assert response_duplicate.status_code 400 assert 已存在 in response_duplicate.json()[detail]6. 常见问题、决策权衡与进阶思考6.1 何时该用何时不该用整洁架构带来了清晰的结构和可维护性但同时也引入了额外的抽象层和复杂度。它并非银弹。适合使用的场景中大型长期项目项目生命周期长功能会持续迭代团队成员可能变动。领域逻辑复杂业务规则多变且核心需要频繁修改和测试。需要多端适配同一套核心业务逻辑需要同时支持Web API、CLI、消息队列消费者等。对测试覆盖率要求高需要高度可测试的代码来保证质量。可能过度设计的场景简单的CRUD管理后台业务逻辑极其简单主要是数据的增删改查。一次性脚本或原型验证快速验证想法不需要长期维护。微服务中的简单数据聚合服务如果服务本身没有复杂逻辑只是调用其他服务并组合数据。我的经验法则当你在一个简单的FastAPI应用中开始感觉路由函数过于庞大或者写单元测试时需要Mock太多东西时就是引入分层架构的好时机。你可以从最核心、最复杂的模块开始实践而不是一次性重写整个项目。6.2 实践中的常见困惑与决策DTO泛滥问题你可能会有CreateUserInput、UpdateUserInput、UserResponse、UserDetailResponse等多个DTO。这虽然精确但也繁琐。一个折中方案是在接口简单时复用或合并一些DTO在接口复杂或需求明确变化时再严格分离。关键在于保持DTO的单一职责避免一个DTO用于多个不同变更频率的用途。实体与ORM模型转换的“胶水代码”在仓储实现中手动转换Entity和Model确实有些冗余。可以使用像maper这样的轻量级库来简化映射或者更激进一点在非常简单的场景下让实体直接继承ORM模型的基类但这会污染领域层。我个人的选择是忍受这部分“胶水代码”因为它明确地标定了架构的边界其成本远低于边界模糊带来的维护代价。依赖注入的复杂度手动管理依赖如我们上面的dependencies.py在项目变大后会变得麻烦。这时可以考虑引入轻量级的DI容器库如dependency-injector或injector来集中管理依赖项的创建和生命周期。但切记不要为了用框架而用框架简单的项目手动管理就足够了。事务管理放在哪一层这是一个经典问题。事务通常与一个完整的“用例”相关例如“创建订单”需要同时操作订单表和库存表。将事务放在基础设施层的仓储中太细粒度放在应用层的用例中比较合适。我们的示例中在get_db_session依赖中通过commit和rollback管理事务这实际上将事务边界定义在了一个HTTP请求内。对于跨多个用例或需要更精细控制的事务可以在用例方法内部显式管理会话。6.3 从模板到实战如何在自己的项目中落地不要试图一次性完美实现所有理论。我建议采用渐进式策略从核心领域开始识别出你系统中最复杂、最核心的业务概念如电商的Order、Payment先为它们建立domain/entities和domain/repositories。实现一个端到端功能挑选一个完整的用户故事如“用户注册”按照分层架构实现从实体、仓储接口、用例、具体仓储到控制器的完整流程。这能帮你打通所有环节理解数据流。建立团队共识确保团队所有成员理解分层的目的和依赖规则。代码审查时要特别注意是否有违反依赖方向如领域层导入sqlalchemy的情况。完善基础设施随着功能增加逐步完善infrastructure层加入缓存Redis、消息队列、文件存储等外部服务的适配器。迭代优化根据项目实际情况调整各层的粒度和职责。架构是服务于项目和团队的没有绝对的标准答案。Flaiers/fastapi-clean-architecture这个项目模板提供了一个优秀的起点和参考。但它更像是一份“地图”而非必须严格遵守的“法律”。理解其背后的原则——分离关注点、依赖倒置、面向接口编程——比照搬其目录结构更重要。在实际项目中灵活运用这些原则构建出适合自己团队和业务节奏的、整洁且高效的后端架构才是我们追求的终极目标。