关于《Clean Architecture》缺失章节的FastAPI+SQLAlchemy架构疑问

云原生阿岚

2026-7-4

Clean Architecture结合FastAPI+SQLAlchemy的架构疑问解答

我刚阅读了Robert Martin所著的《Clean Architecture》，针对其中的《The Missing Chapter》产生了一些疑问。在使用FastAPI+SQLAlchemy时，我们会编写数据库初始化相关代码，示例如下：

from logging import getLogger
from typing import Any, Callable, Optional

import orjson
from app.config import PROJECT_NAME, settings
from sqlalchemy import MetaData
from sqlalchemy.engine.url import URL
from sqlalchemy.ext import asyncio as sa_asyncio
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlalchemy.orm import sessionmaker

logger = getLogger(__name__)


def get_master_dsn(is_test: bool = False):
    master_dsn = URL.create(
        drivername=settings.DB.PROTOCOL,
        username=settings.DB.USER,
        password=settings.DB.PASSWORD,
        host=settings.DB.HOST,
        port=settings.DB.PORT,
        database=settings.DB.DATABASE,
    ).render_as_string(hide_password=False)

    if is_test:
        master_dsn += "_test"

    return master_dsn


POSTGRESQL_MASTER_DSN = get_master_dsn()


def orjson_dumps(value: Any, *, default: Optional[Callable[[Any], Any]]) -> str:
    return orjson.dumps(value, default=default).decode()


def init_db(long_operation: bool = False, is_test: bool = False) -> AsyncEngine:
    dsn = get_master_dsn(is_test)

    timeout_command = (
        settings.DB.TIMEOUT_COMMAND * 100
        if long_operation
        else settings.DB.TIMEOUT_COMMAND
    )

    return sa_asyncio.create_async_engine(
        dsn,
        max_overflow=0,
        connect_args={
            "server_settings": {"application_name": PROJECT_NAME, "timezone": "utc"},
            "timeout": settings.DB.TIMEOUT_CONNECTION,
            "command_timeout": timeout_command,
        },
        pool_pre_ping=True,
        json_serializer=orjson_dumps,
        json_deserializer=orjson.loads,
    )


db = init_db()
Session = sessionmaker(bind=db, class_=sa_asyncio.AsyncSession, expire_on_commit=False)
metadata = MetaData(schema=settings.DB.SCHEMA)
Base = declarative_base(metadata=metadata)

参考书中的图36.4组件布局，我的疑问如下：

疑问列表

数据库准备相关代码应放置在该布局的哪个位置？
SQLAlchemy模型是否应放在组件内的models.py中？
接口应如何存储？
如下所示的项目结构是否合理？

web
components
- orders
-- usecases
-- models 
-- utils
- another_component
....

另外，书中第26章提到的Main组件，是否是放置相关代码的合适位置？

问题解答

1. 数据库准备代码的放置位置

数据库初始化（engine、sessionmaker、Base声明）这类代码属于基础设施层，负责对接外部资源（数据库），应该放在独立的infrastructure/db/目录下，不与业务组件耦合。参考图36.4的布局，它不属于任何业务组件，是位于组件外部的支撑性代码。

如果用书中Main组件的思路，Main作为顶层协调组件，可以负责触发数据库初始化的启动逻辑，但具体的配置、engine创建等细节仍要放在基础设施层，Main只做依赖注入和启动协调的工作。

2. SQLAlchemy模型的存放位置

SQLAlchemy模型是业务核心实体的数据库映射，属于领域层内容，应该放在对应业务组件内部的models/目录下（比如components/orders/models/）。每个业务组件的模型仅服务于自身业务，这样能保证组件的独立性，符合Clean Architecture“实体位于最内层”的原则。

如果采用Repository模式，模型是基础设施层的映射实现，领域模型用纯Python类，但即便如此，模型也应该和对应业务组件绑定，避免跨组件依赖。

3. 接口的存储方式

接口分两种场景处理：

业务用例接口：定义业务组件对外提供的能力，属于应用层，应该放在业务组件内的usecases/目录下（比如components/orders/usecases/interfaces.py），只规定输入输出和业务规则，不依赖具体实现。
FastAPI API接口：属于适配器层，负责将业务能力暴露为HTTP接口，应该放在独立的web/或adapters/api/目录下。API路由仅调用业务用例，不包含业务逻辑，确保业务组件不依赖API层，符合依赖倒置原则。

4. 给定项目结构的合理性

你给出的结构整体方向正确，但可以做以下优化：

新增独立的infrastructure/目录，存放数据库、缓存等基础设施代码，避免与业务组件混放；
每个业务组件内明确分层，比如orders/application/usecases/（应用层用例）、orders/domain/models/（领域实体）、orders/infrastructure/repositories/（组件内的数据库适配实现）；
web/目录专注作为API适配器层，仅存放FastAPI路由、请求响应模型等内容。

关于`Main`组件的补充

Main组件是顶层的应用入口协调者，负责组装所有依赖——比如初始化基础设施、注入业务组件的依赖、启动API服务等。它不属于任何业务组件或基础设施层，位于架构最外层。数据库初始化的启动逻辑可以放在Main里，但具体实现细节仍要放在infrastructure/db/中，Main仅做调用协调。

内容的提问来源于stack exchange，提问作者Альберт Александров