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( username=input_dto.username, email=input_dto.email, hashed_password=hashed_password, is_active=True ) # 调用仓储接口保存，不关心是存到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的职责非常单一。它接收一个输入DTO（Data Transfer Object），执行业务规则（如唯一性检查），操作领域实体，并通过抽象的仓储接口进行持久化。它不包含任何HTTP状态码、JSON序列化或SQL语句。

应用层DTO(create_user_input.py) 是进入应用层的数据契约，通常也很简单：

from pydantic import BaseModel, EmailStr class CreateUserInput(BaseModel): username: str email: EmailStr password: str

3.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( id=model.id, username=model.username, email=model.email, hashed_password=model.hashed_password, is_active=model.is_active, created_at=model.created_at ) def _to_model(self, entity: User) -> UserModel: """将领域实体转换为数据库模型（用于保存）""" return UserModel( id=entity.id, username=entity.username, email=entity.email, hashed_password=entity.hashed_password, is_active=entity.is_active, created_at=entity.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_code=status.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_repo=user_repo) try: created_user = await use_case.execute(user_input) except ValueError as e: # 捕获用例中的业务逻辑异常（如用户名重复） raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=str(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(session=session) # 可以继续定义其他仓储或服务的依赖项 # 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(title=settings.PROJECT_NAME, version=settings.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, host="0.0.0.0", port=8000)

这个文件非常干净，它的职责就是创建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(username="alice123", email="alice@example.com", hashed_password="hash") assert user.username == "alice123" assert user.is_active is True # 测试默认值 def test_user_username_validation(): """测试用户名字段的自定义验证器""" # 有效用户名 User(username="alice123", email="a@b.com", hashed_password="hash") # 用户名过短 with pytest.raises(ValueError, match="用户名至少3个字符"): User(username="ab", email="a@b.com", hashed_password="hash") # 用户名包含非法字符 with pytest.raises(ValueError, match="用户名只能包含字母和数字"): User(username="alice-123", email="a@b.com", hashed_password="hash") def test_user_business_methods(): """测试实体的业务方法""" user = User(username="alice", email="a@b.com", hashed_password="old_hash", is_active=False) user.activate() assert user.is_active is True user.change_password("new_hash") assert user.hashed_password == "new_hash"

5.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(id=1, username="alice", email="a@b.com", hashed_password="hashed_pw") mock_repo.save.return_value = saved_user # 2. 实例化用例，注入模拟仓储 use_case = CreateUserUseCase(user_repo=mock_repo) input_dto = CreateUserInput(username="alice", email="a@b.com", password="secret123") # 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(id=99, username="alice", email="existing@b.com", hashed_password="hash") mock_repo.get_by_username.return_value = existing_user use_case = CreateUserUseCase(user_repo=mock_repo) input_dto = CreateUserInput(username="alice", email="new@b.com", password="secret123") 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 = "sqlite+aiosqlite:///:memory:" test_engine = create_async_engine(TEST_DATABASE_URL, echo=False) TestingSessionLocal = sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) @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": "test@example.com", "password": "strongpassword" } response = test_client.post("/users/", json=user_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/", json=user_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这个项目模板提供了一个优秀的起点和参考。但它更像是一份“地图”，而非必须严格遵守的“法律”。理解其背后的原则——分离关注点、依赖倒置、面向接口编程——比照搬其目录结构更重要。在实际项目中，灵活运用这些原则，构建出适合自己团队和业务节奏的、整洁且高效的后端架构，才是我们追求的终极目标。