1. 项目概述为什么我们需要一个“干净”的FastAPI后端如果你和我一样用FastAPI写过几个项目从简单的API服务到稍复杂些的业务系统你可能会发现一个现象项目初期代码结构清晰逻辑一目了然。但随着需求迭代、功能堆叠main.py文件越来越臃肿数据库模型、业务逻辑、路由定义、依赖注入、工具函数全都搅在一起。想改个业务规则得在好几个文件里跳来跳去想加个新功能得小心翼翼生怕破坏了原有的脆弱平衡。这种代码我们通常称之为“面条式代码”或“大泥球架构”。“Flaiers/fastapi-clean-architecture”这个项目就是针对这个痛点的一剂良药。它不是一个要你照搬的“终极框架”而是一个基于“整洁架构”思想的FastAPI项目脚手架和最佳实践范例。它的核心目标是帮你从一开始就建立起一个边界清晰、职责分离、易于测试和维护的代码结构。简单说它教你如何用FastAPI写出“专业级”的后端代码而不是停留在快速验证想法的原型阶段。这个项目特别适合两类开发者一是已经熟悉FastAPI基础但苦于项目结构混乱希望提升代码质量的进阶者二是正准备启动一个中长期、业务逻辑可能比较复杂的新项目希望有一个坚实起点的团队。它把那些在大型企业级应用中反复验证过的架构思想以一种清晰、可操作的方式带到了FastAPI这个现代、高效的生态中。2. 整洁架构核心思想与FastAPI的融合之道2.1 什么是整洁架构它解决了什么问题整洁架构也叫“洋葱架构”或“端口与适配器架构”是由Robert C. MartinUncle Bob提出的一套软件设计原则。它的核心画像是一个同心圆从内到外分别是实体Entities、用例Use Cases、接口适配器Interface Adapters和框架与驱动Frameworks Drivers。这个结构的关键在于依赖关系规则内层圆更核心的代码不应该知道外层圆更具体的任何事情。也就是说业务逻辑实体和用例不应该依赖于Web框架FastAPI、数据库SQLAlchemy或者任何外部服务。所有依赖的方向都是指向圆心的。这带来的最大好处是可替换性和可测试性。你的核心业务规则是独立且稳定的今天用FastAPI明天换另一个Web框架今天用PostgreSQL明天换MongoDB核心逻辑几乎不需要改动。测试时你可以轻松地用内存中的模拟对象替换掉真实的数据层实现快速、独立的单元测试。那么FastAPI作为一个高性能的现代Web框架如何与这套略显“学院派”的理论结合呢这正是“fastapi-clean-architecture”项目的精妙之处。它没有生硬地套用理论而是做了非常务实的落地。2.2 项目中的分层设计与职责划分该项目通常会将代码组织成以下几个核心目录每一层都有明确的职责domain/(领域层): 这是最内层、最纯粹的部分。这里放置的是与任何技术实现无关的业务核心。主要包括实体Entities: 用普通的Python类或Pydantic模型定义业务对象如User、Product、Order。它们只有属性和最基础的方法如验证没有数据库字段映射。值对象Value Objects: 不可变的、描述事物特征的对象如EmailAddress、Money。仓储抽象Repository Interfaces: 这里定义的是接口Python中的Protocol或ABC抽象基类例如IUserRepository。它只声明了“我需要哪些数据操作”如save,get_by_id,find_by_email但绝不包含具体实现比如SQL语句。这确保了业务逻辑只依赖于抽象而非具体的数据技术。application/(应用层/用例层): 这一层包含具体的用例Use Cases或服务Services。每个用例代表一个完整的用户操作比如“用户注册”、“创建订单”。它负责协调领域实体、执行业务规则、并调用仓储接口来持久化数据。关键点应用层服务只通过接口调用仓储它不知道数据是存在MySQL里还是Redis里。同时它也不处理HTTP请求的细节如状态码、Cookie只返回领域对象或简单的DTO。infrastructure/(基础设施层): 这里是所有外部依赖的具体实现。它依赖于内层实现了内层定义的接口但内层对它一无所知。主要包括仓储实现Repository Implementations: 例如SQLAlchemyUserRepository它实现了IUserRepository接口内部使用SQLAlchemy的Session和Model来操作数据库。数据库模型ORM Models: SQLAlchemy或Tortoise-ORM的模型定义在这里它们负责与数据库表的映射。外部服务客户端: 调用第三方API如支付、短信的具体代码。配置、日志、邮件发送等具体实现。api/(接口适配器层): 这一层是Web世界与业务世界的翻译官。在FastAPI项目中它主要包含路由Routers: FastAPI的APIRouter定义在这里负责将HTTP请求映射到具体的应用层服务调用。依赖注入Dependencies: 定义如何获取当前用户、数据库会话等并将其“注入”到路由处理函数中。请求/响应模型Schemas: 使用Pydantic定义API的输入和输出格式。它们负责将JSON数据转换为应用层能理解的对象并将应用层返回的结果序列化为JSON。注意这里的Schema和领域层的Entity是分开的因为API的字段可能只是实体的一部分或者需要做格式转换。core/(核心配置与共享): 放置项目配置、依赖注入容器设置、异常处理、中间件等跨层共享的组件。提示这种分层不是死板的教条。对于非常简单的CRUD操作你可能觉得“杀鸡用牛刀”。但项目的价值在于当业务逻辑变得复杂如订单涉及库存锁定、优惠计算、支付触发、消息通知等多个步骤时这种结构能强制你进行清晰的思考和解耦避免代码腐化。3. 从零开始基于此理念构建一个用户模块理论说再多不如动手写一行代码。让我们以最常见的“用户注册”功能为例看看如何按照这个架构一步步实现。3.1 领域层定义业务核心首先我们在domain/entities/user.py中定义用户实体。注意它没有id的默认值因为ID的生成策略自增、UUID属于基础设施层的职责。# domain/entities/user.py from pydantic import BaseModel, EmailStr, field_validator from datetime import datetime from typing import Optional class User(BaseModel): 用户领域实体 id: Optional[int] None # ID由数据库或服务生成 email: EmailStr username: str hashed_password: str # 存储的是哈希后的密码明文密码不应出现在领域层 is_active: bool True created_at: Optional[datetime] None updated_at: Optional[datetime] None field_validator(username) def validate_username(cls, v): if len(v) 3: raise ValueError(用户名至少3个字符) if not v.isalnum(): raise ValueError(用户名只能包含字母和数字) return v def activate(self): 激活用户业务逻辑 self.is_active True self.updated_at datetime.utcnow() def deactivate(self): 停用用户业务逻辑 self.is_active False self.updated_at datetime.utcnow()接着在domain/repositories/user_repository.py中定义仓储接口。我们使用typing.Protocol来声明这样任何实现了这些方法的类都可以被当作仓储使用。# domain/repositories/user_repository.py from typing import Protocol, Optional from domain.entities.user import User class IUserRepository(Protocol): 用户仓储接口端口 async def get_by_id(self, user_id: int) - Optional[User]: ... async def get_by_email(self, email: str) - Optional[User]: ... async def create(self, user: User) - User: ... async def update(self, user: User) - User: ... async def delete(self, user_id: int) - bool: ...3.2 应用层实现用户注册用例现在在application/services/user_service.py中实现具体的业务逻辑。服务会依赖仓储接口并通过依赖注入的方式获得具体的仓储实现。# application/services/user_service.py from typing import Optional from domain.entities.user import User from domain.repositories.user_repository import IUserRepository from passlib.context import CryptContext pwd_context CryptContext(schemes[bcrypt], deprecatedauto) class UserService: def __init__(self, user_repo: IUserRepository): # 依赖注入仓储接口而不是具体实现 self.user_repo user_repo async def register_user(self, email: str, username: str, password: str) - User: 用户注册用例 # 1. 检查邮箱是否已存在业务规则 existing_user await self.user_repo.get_by_email(email) if existing_user: raise ValueError(该邮箱已被注册) # 2. 创建用户实体核心业务对象创建 hashed_password pwd_context.hash(password) user User( emailemail, usernameusername, hashed_passwordhashed_password, created_atdatetime.utcnow() ) # 3. 持久化用户通过接口调用不关心具体数据库 created_user await self.user_repo.create(user) return created_user async def authenticate_user(self, email: str, password: str) - Optional[User]: 用户认证用例 user await self.user_repo.get_by_email(email) if not user or not pwd_context.verify(password, user.hashed_password): return None if not user.is_active: raise ValueError(用户账户已被停用) return user注意这里UserService只接收一个IUserRepository类型的参数。至于运行时传入的是操作MySQL的仓储还是操作Redis的仓储服务本身不关心。这为单元测试提供了极大便利你可以传入一个模拟的、内存中的仓储实现。3.3 基础设施层实现SQLAlchemy仓储与模型现在我们需要为之前定义的接口提供一个具体的实现。在infrastructure/repositories/sqlalchemy_user_repository.py中# infrastructure/repositories/sqlalchemy_user_repository.py from typing import Optional from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from domain.entities.user import User from domain.repositories.user_repository import IUserRepository from infrastructure.models.user import UserModel # SQLAlchemy的ORM模型 class SQLAlchemyUserRepository(IUserRepository): IUserRepository的SQLAlchemy实现适配器 def __init__(self, session: AsyncSession): self.session session async def get_by_id(self, user_id: int) - Optional[User]: result await self.session.execute( select(UserModel).where(UserModel.id user_id) ) user_model result.scalar_one_or_none() return self._to_entity(user_model) if user_model else None async def get_by_email(self, email: str) - Optional[User]: result await self.session.execute( select(UserModel).where(UserModel.email email) ) user_model result.scalar_one_or_none() return self._to_entity(user_model) if user_model else None async def create(self, user: User) - User: user_dict user.model_dump(exclude{id}) user_model UserModel(**user_dict) self.session.add(user_model) await self.session.flush() # 获取生成的ID await self.session.refresh(user_model) return self._to_entity(user_model) def _to_entity(self, model: UserModel) - User: 将数据库模型转换为领域实体 return User( idmodel.id, emailmodel.email, usernamemodel.username, hashed_passwordmodel.hashed_password, is_activemodel.is_active, created_atmodel.created_at, updated_atmodel.updated_at ) # ... 实现update和delete方法对应的SQLAlchemy模型定义在infrastructure/models/user.py# infrastructure/models/user.py from sqlalchemy import Column, Integer, String, Boolean, DateTime from sqlalchemy.sql import func from infrastructure.database import Base # 假设有一个Base类 class UserModel(Base): __tablename__ users id Column(Integer, primary_keyTrue, indexTrue) email Column(String(255), uniqueTrue, indexTrue, nullableFalse) username Column(String(50), uniqueTrue, indexTrue, nullableFalse) hashed_password Column(String(255), nullableFalse) is_active Column(Boolean, defaultTrue) created_at Column(DateTime(timezoneTrue), server_defaultfunc.now()) updated_at Column(DateTime(timezoneTrue), onupdatefunc.now())3.4 接口层构建FastAPI路由与依赖最后我们将所有部分连接起来暴露为HTTP API。在api/v1/endpoints/users.py中# api/v1/endpoints/users.py from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from typing import Any from api.v1.schemas.user import UserCreate, UserOut # Pydantic请求/响应模型 from application.services.user_service import UserService from infrastructure.repositories.sqlalchemy_user_repository import SQLAlchemyUserRepository from core.deps import get_db # 获取数据库会话的依赖 router APIRouter() router.post(/register, response_modelUserOut, status_codestatus.HTTP_201_CREATED) async def register_user( *, db: AsyncSession Depends(get_db), # 注入数据库会话 user_in: UserCreate, # 使用Pydantic模型验证输入 ) - Any: 用户注册 # 1. 组装基础设施层组件 user_repo SQLAlchemyUserRepository(sessiondb) # 2. 注入到应用层服务 user_service UserService(user_repouser_repo) try: # 3. 调用业务用例 user_entity await user_service.register_user( emailuser_in.email, usernameuser_in.username, passworduser_in.password ) except ValueError as e: # 4. 将业务异常转换为HTTP异常 raise HTTPException( status_codestatus.HTTP_400_BAD_REQUEST, detailstr(e) ) # 5. 将领域实体转换为API响应模型 return UserOut.from_entity(user_entity)对应的请求/响应模型在api/v1/schemas/user.py中它们负责API边界的序列化# api/v1/schemas/user.py from pydantic import BaseModel, EmailStr from datetime import datetime from domain.entities.user import User class UserCreate(BaseModel): 用户注册请求模型 email: EmailStr username: str password: str class Config: from_attributes True class UserOut(BaseModel): 用户信息响应模型 id: int email: EmailStr username: str is_active: bool created_at: datetime classmethod def from_entity(cls, user: User): 从领域实体转换 return cls( iduser.id, emailuser.email, usernameuser.username, is_activeuser.is_active, created_atuser.created_at )3.5 依赖注入容器的配置为了让依赖管理更清晰项目通常会使用一个依赖注入容器如dependency-injector库或在core/deps.py中手动管理。这里展示一个简单的手动方式# core/deps.py from typing import AsyncGenerator from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from core.config import settings # 从配置读取数据库URL engine create_async_engine(settings.DATABASE_URL, echoTrue) AsyncSessionLocal async_sessionmaker(engine, expire_on_commitFalse) async def get_db() - AsyncGenerator[AsyncSession, None]: 获取数据库会话的依赖 async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception: await session.rollback() raise finally: await session.close() # 可以在这里定义获取服务的快捷函数 def get_user_service(db: AsyncSession Depends(get_db)): user_repo SQLAlchemyUserRepository(sessiondb) return UserService(user_repouser_repo)这样在路由中就可以直接使用user_service: UserService Depends(get_user_service)来注入一个配置好的服务实例。4. 项目实战测试策略与部署考量采用整洁架构后测试变得异常清晰和简单。因为各层之间通过接口解耦你可以针对每一层进行独立的、专注的测试。4.1 分层测试策略领域层单元测试测试实体类的方法和验证逻辑。这部分测试最快不依赖任何外部资源。# tests/domain/test_user.py def test_user_activation(): user User(emailtestexample.com, usernametest, hashed_passwordhash) user.deactivate() assert user.is_active is False user.activate() assert user.is_active is True assert user.updated_at is not None应用层单元测试测试服务用例的逻辑。这里可以使用unittest.mock来模拟仓储接口完全隔离数据库。# tests/application/test_user_service.py import pytest from unittest.mock import AsyncMock from application.services.user_service import UserService pytest.mark.asyncio async def test_register_user_success(): # 1. 创建模拟仓储 mock_repo AsyncMock() mock_repo.get_by_email.return_value None # 模拟邮箱不存在 mock_repo.create.return_value User(id1, emailab.com, ...) # 模拟创建成功 # 2. 实例化服务注入模拟仓储 service UserService(user_repomock_repo) # 3. 执行测试 user await service.register_user(ab.com, user1, password123) # 4. 断言 assert user.id 1 mock_repo.get_by_email.assert_called_once_with(ab.com) mock_repo.create.assert_called_once()基础设施层集成测试测试具体的仓储实现是否与数据库正确交互。需要启动一个测试数据库如使用pytest-asyncio 测试容器。# tests/infrastructure/test_repositories.py pytest.mark.asyncio async def test_sqlalchemy_user_repository_create(session): # session是测试用的数据库会话 repo SQLAlchemyUserRepository(session) user_entity User(emailtestrepo.com, usernamerepo, hashed_passwordhash) saved_user await repo.create(user_entity) assert saved_user.id is not None assert saved_user.email testrepo.comAPI端到端测试使用TestClient测试完整的HTTP请求-响应流程。这部分测试速度较慢但能验证整个链路是否通畅。# tests/api/test_users.py from fastapi.testclient import TestClient from main import app # 你的FastAPI应用实例 client TestClient(app) def test_register_user_endpoint(): response client.post(/api/v1/users/register, json{ email: e2etest.com, username: e2euser, password: e2epass }) assert response.status_code 201 data response.json() assert data[email] e2etest.com4.2 部署与配置管理一个结构清晰的项目配置管理也应如此。建议使用Pydantic的BaseSettings来管理配置区分不同环境开发、测试、生产。# core/config.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): PROJECT_NAME: str My Clean FastAPI Project API_V1_STR: str /api/v1 DATABASE_URL: str SECRET_KEY: str ALGORITHM: str HS256 ACCESS_TOKEN_EXPIRE_MINUTES: int 30 class Config: env_file .env # 从.env文件加载配置 case_sensitive True settings Settings()在部署时通过环境变量注入DATABASE_URL和SECRET_KEY等敏感信息。使用Docker容器化部署是推荐的做法Dockerfile和docker-compose.yml的编写也会因为项目结构清晰而变得简单。5. 常见陷阱、决策点与进阶技巧在实际采用这种架构时你会遇到一些抉择和坑点。以下是我从经验中总结的一些关键点。5.1 何时需要引入整洁架构不要为了架构而架构。对于生命周期很短小于3个月、业务逻辑极其简单的原型或内部工具传统的“单片”FastAPI应用所有代码放在几个文件里可能更高效。整洁架构的收益在项目中期和后期才会爆发式体现它需要前期投入更多的设计时间。一个简单的判断标准是如果你的业务逻辑包含了大量的if-else规则、需要频繁与不同外部服务交互、或者你预见到团队会扩大、功能会持续复杂化那么从一开始就采用整洁架构是明智的。5.2 领域实体与数据库模型的映射之痛这是最常见的困惑点之一。我们有了User实体又有了UserModelORM模型感觉像是重复劳动。关键在于理解它们的职责不同实体承载业务规则和行为如user.activate()。它应该是“贫血”的吗在DDD领域驱动设计中实体应该是“充血”的包含业务方法。但在许多实践中一个只包含数据验证的“贫血模型”在Python中也很常见业务逻辑放在服务里。项目通常采用折中方案。ORM模型只关心如何高效地持久化和查询数据。它包含数据库特定的细节如索引、关系、字段类型映射。映射代码如_to_entity看起来是样板代码但这是必要的“转换层”。对于简单字段可以使用像model_dump和**dict这样的快捷方式。对于复杂映射可以考虑使用专门的库如pydantic的model_validate或mapper库但手动映射通常更清晰、可控。5.3 依赖注入的实践手动 vs 容器上面的例子我们使用了FastAPI自带的Depends进行手动依赖注入。对于中小型项目这完全足够且直观。但当依赖关系变得非常复杂例如一个服务依赖多个仓储仓储又依赖不同的数据库连接池和配置时手动管理会变得繁琐。此时可以考虑引入依赖注入容器如dependency-injector或injector。它们允许你在一个中心位置通常是core/containers.py声明所有组件的生命周期和依赖关系然后在需要的地方自动注入。这提升了可维护性但也增加了学习成本和项目复杂度。我的建议是从简单的Depends开始当感到疼痛时再引入容器。5.4 事务管理放在哪一层这是一个有争议的话题。严格遵循整洁架构事务边界应该由最外层的“适配器”来控制因为事务是基础设施数据库的细节。在FastAPI中通常会在路由层或一个中间件开启数据库事务在请求成功完成后提交异常时回滚。正如我们在get_db依赖中做的那样。这意味着一个HTTP请求对应一个事务。如果单个用例内需要操作多个聚合根实体它们会在同一个事务中。如果业务规则要求跨多个用例保持一致性则需要引入更复杂的模式如领域事件Domain Events配合最终一致性Eventual Consistency这属于更高级的领域驱动设计范畴。5.5 性能考量与优化分层带来了清晰度也可能引入微小的性能开销主要是对象转换和额外的调用。对于绝大多数Web应用这点开销与网络I/O和数据库查询相比微不足道。不要过早优化。真正的性能瓶颈往往出现在N1查询问题在应用层循环调用仓储的单个查询方法。解决方案是让仓储接口提供批量查询或包含复杂关联查询的方法如get_users_with_orders在基础设施层通过SQL JOIN一次性完成。对象转换开销对于返回大量数据的列表接口逐行转换ORM模型到实体再到响应模型可能较慢。可以考虑在仓储层直接返回字典或特定的DTO或者对于只读场景允许响应模型直接从ORM模型构造如果字段匹配但这会轻微破坏架构纯洁性需要权衡。我个人在实际项目中对于简单的CRUD列表有时会允许Pydantic响应模型直接从SQLAlchemy模型model_dump并在文档中注明这是出于性能考虑的妥协。对于核心的写操作和复杂业务逻辑则严格遵守分层转换。采用“Flaiers/fastapi-clean-architecture”所倡导的架构模式初期你会觉得束缚多写了很多“看似无用”的接口和转换代码。但当你需要修改数据库、需要为某个功能编写全覆盖的单元测试、或者需要理清一段复杂的业务逻辑时你会感谢当初这些“束缚”带来的清晰边界。它迫使你思考每个对象的职责、每个依赖的方向最终得到的是一套坚韧、适应性强、能让团队高效协作的代码基底。这不仅仅是代码结构的变化更是一种构建可维护软件思维方式的训练。