博客地址: https://www.thedigitalcatbooks.com/pycabook-introduction/
《TDD in Python with pytest》详见 TDD/readme.md
作者设定了一个简单的 Web 应用场景:用户访问一个 URL(例如 /rooms?status=available)来查看可租用的房间。系统需要处理这个请求并返回结果。
博客通过数据在系统中流动的过程,引出了几个关键层级:
- Web 框架层(Web Framework):
- 职责: 处理 HTTP 协议。它负责解码 HTTP 请求,提取参数(如
status=available),并将信息传递给内部逻辑。 - 原则: 它不应该包含业务逻辑。它的领域仅限于 HTTP 协议。
- 业务逻辑层 / 用例(Use Case):
- 职责: 系统的核心。它实现了具体的业务规则(例如:如何过滤和排序房间)。
- 地位: 用例是系统中最重要、最核心的部分,代表了软件存在的真正价值。
- 存储系统层(Storage System):
- 职责: 提供数据。它可以是数据库(如 PostgreSQL)、文件、外部 API 或传感器。
- 地位: 被视为实现细节(Implementation Detail)。
博客深入解释了为什么需要这种分层架构:
- 关注点分离(Separation of Concerns):
- 不同的组件负责不同的任务。Web 框架管通信,存储层管数据持久化,用例管业务逻辑。
- 好处: 降低耦合,使系统更易于测试、维护和更换组件。
- 抽象(Abstraction)与实现细节(Implementation Detail):
- 业务逻辑(用例)不应该依赖于具体的数据库语法或特定的 Web 框架。
- 具体的数据库(如 Postgres)或 Web 协议被视为“细节”。核心逻辑应保持纯净,不受细节变动的影响。
- 控制反转(Inversion of Control, IoC):
- 问题: 如果用例直接调用数据库 API,两者就会强耦合。
- 解决方法:
- 接口(Interface): 为存储系统定义一个标准接口(如
list_rooms_with_status)。 - 依赖注入: 用例在初始化时接收一个符合接口要求的实例,而不是自己去实例化一个具体的数据库对象。
- 灵活性(Flexibility): 你可以轻松地将 Web 界面替换为命令行界面(CLI),或者将 SQL 数据库替换为 JSON 文件存储,而不需要修改核心业务逻辑。
- 可测试性(Testability): 由于层级之间解耦,你可以独立地测试每一层。例如,在测试 Web 层时,可以使用一个“伪造的用例”(Mock Use Case)来模拟业务逻辑,从而只验证 HTTP 处理是否正确。
整洁架构的核心在于保护业务逻辑(用例),使其不与外部工具(Web、数据库)直接绑定,从而构建一个健壮、灵活且易于测试的系统。
作者指出,一个优秀的软件系统应该像一个分工明确的工厂。为了实现控制和降低复杂度,必须将系统划分为不同的子系统,并建立严谨的边界。
- 责任明确:每个组件应明确自己的职责,避免重叠导致冲突或低效。
- 处理耦合:数据类型和格式是耦合的一种形式。架构设计的关键在于确定哪个部分定义“语言”(数据格式),并确保依赖关系不会破坏系统核心。
整洁架构被描绘为一个圆形的层级结构,从内到外分别是:
- 实体层 (Entities - 最内层):
- 内容:代表领域模型(Domain Models),是业务中最基础的概念。
- 特点:在 Python 中通常是简单的类(轻量级模型)。它们不连接数据库、不包含框架代码(如 Django Models)、不处理 JSON 序列化。
- 依赖:实体层对外部一无所知,是整个系统的基石。
- 用例层 (Use Cases):
- 内容:实现业务逻辑(Business Rules)。例如:“用户登录”、“搜索过滤器”、“银行转账”。
- 特点:用例应尽量保持细小且单一,以便测试。
- 依赖:可以访问实体,也可以调用其他用例,但对外部的网关或系统无感知。
- 网关层 (Gateways):
- 内容:定义与外部系统(如数据库、第三方服务)交互的接口(API)。
- 特点:它隐藏了外部系统的实现细节。比如,网关定义了“如何保存数据”,但不关心是用 PostgreSQL 还是 MongoDB。
- 外部系统层 (External Systems - 最外层):
- 内容:具体的实现细节,包括 Web 框架(Django, Flask)、数据库驱动、命令行界面等。
- 角色:它们通常是“触发者”,通过调用用例来执行业务逻辑。
- 向内调用(Talk Inwards)使用简单结构:当外层需要调用内层时,直接使用简单的对象(如实体或 Python 原生类型)。
- 向外调用(Talk Outwards)通过接口:当内层(如用例)需要访问外层功能(如存入数据库)时,必须通过网关层定义的接口,而不能直接依赖具体的实现。
- 抽象级别:越向内越抽象(业务概念),越向外越具体(实现细节)。
- 灰色地带:架构并非死板的教条。在某些情况下(如性能极度敏感时),可能需要打破规则直接访问外层 API,但必须在代码中添加显眼警告。
- 一致性:如果频繁需要打破层级,说明架构分层可能过细,应考虑合并层级。
本章建立了一套“依赖向内”的层级秩序,确保了核心业务逻辑(用例和实体)与外部工具(数据库和框架)的解耦。这种设计使得系统更易于测试、维护,并能在不改变业务逻辑的情况下轻松替换外部插件。
- 目标:创建一个简单的搜索引擎,允许用户通过过滤器(如尺寸、价格、经纬度)搜索房间。
- 核心对象:
Room(房间),包含唯一标识符(code)、大小、价格、纬度和经度。 - 核心理念:通过极简的业务逻辑演示如何分离系统层级。
- 实现方式:使用 Python 的
dataclasses来定义Room类。 - 特点:模型非常轻量,仅包含数据和基础的转换方法(如
from_dict和to_dict)。 - TDD(测试驱动开发):作者强调先写测试(
test_room_model_init等),确保模型可以正确初始化、比较以及与字典格式相互转换。
- 目的:将领域模型转换为语言无关的格式(如 JSON),以便通过 API 返回。
- 实现:编写了一个
RoomJsonEncoder(继承自json.JSONEncoder),处理UUID等非原生 JSON 类型。 - 架构意义:序列化逻辑不放在领域模型中,而是在外部专门的序列化层,保持模型的纯净。
- 业务逻辑主体:本章实现了一个最简单的用例
room_list_use_case,其功能是获取仓库中所有的房间列表。 - 依赖倒置:用例不直接依赖具体的数据库,而是接受一个“仓库接口”(Repository Interface)。
- 实现形式:作者在这里使用了函数而不是类,因为代码需要维护,所以越简单越好。
- 测试方法:通过
unittest.mock模拟仓库对象,确保用例正确调用了仓库的方法。
- 内存存储 (MemRepo):本章实现了一个临时的内存存储库作为外部系统示例。
- 职责:它接收数据并将其转换为领域模型列表。
- 可插拔性:强调了架构的灵活性,未来可以轻松地将这个内存仓库替换为 Postgres 或 MongoDB。
- 系统集成:创建了一个
cli.py脚本,演示如何将所有部分串联起来:
- 初始化存储库(数据源)。
- 运行用例并将存储库作为参数注入。
- 打印结果。
- 优势:这种结构允许在不改变核心业务逻辑的情况下,随意更换外部接口(如从 CLI 切换到 Web API)。
本章的核心在于展示层级分离:
- 实体 (Entities):
Room类。 - 用例 (Use Cases):
room_list_use_case函数。 - 网关与外部系统 (Gateways & External Systems):
MemRepo。
- 在创建的 tests/ 的每个子目录中创建一个空文件 init.py,以解决pytest 在不同的目录下发现两个同名的测试文件导致的命名冲突
- 使用
pip install -r requirements/dev.txt安装依赖包
在第3章中,作者实现了一个简单的命令行界面(CLI)。本章通过引入 Flask 框架,展示了如何在不改变业务逻辑(Use Case)和数据层(Repository)的情况下,仅通过更换表现层,将系统扩展到 Web 平台。
- Web 框架: 选择轻量级的 Flask。
- 依赖管理: 更新了
requirements/prod.txt(加入 Flask)和requirements/test.txt(加入pytest-flask用于测试)。 - 应用工厂: 创建了
application/app.py,使用工厂模式(App Factory)来初始化 Flask 应用并注册蓝图(Blueprints)。
作者强调即使是 Web 接口也应遵循 TDD。
- 测试工具: 使用
pytest-flask插件,它提供了client固件(Fixture)来模拟 HTTP 请求。 - Mock 机制: 在测试 Web 接口时,Mock(模拟)了 Use Case。因为测试的目标是验证“网关”是否正确调用了业务逻辑并返回了正确的 JSON 格式,而不是再次测试业务逻辑本身。
- 验证点: 测试用例检查了响应的 JSON 内容、HTTP 状态码(200)以及 MIME 类型(application/json)。
在 application/rest/room.py 中定义了 /rooms 路由:
- 解耦体现: 路由处理函数内部的操作与 CLI 界面几乎完全一致:
- 初始化存储库(
MemRepo)。 - 调用用例(
room_list_use_case)。 - 返回响应。
- 序列化: 使用之前章节定义的
RoomJsonEncoder将领域模型对象转换为 JSON 格式。
- 逻辑复用: 业务逻辑(Use Case)完全不知道自己是被 CLI 调用还是被 Flask 调用。
- 可维护性: 系统被划分为不同的层(领域层、用例层、存储层、表现层)。
- 易于测试: 每一层都可以独立测试,外部依赖(如 Web 框架)可以轻松被模拟。
本章通过在 Flask 中复用相同的 Use Case,完美演示了整洁架构如何实现“业务逻辑与外部框架解耦”的核心原则。
在powershell中输入$env:FLASK_CONFIG = "development";uv run flask run启动后端服务,然后点击http://127.0.0.1:5000/rooms查看端点返回的 JSON 数据
在清洁架构中,用例(Use Cases)是逻辑的核心,也是错误处理的主要场所。作者通过引入“请求(Request)”和“响应(Response)”模式,展示了如何构建一个健壮的错误处理机制。
作者指出,错误管理不应直接依赖于具体的框架(如 HTTP 状态码),而应该在用例层通过抽象的请求对象和响应对象来处理。
- 请求对象(Request Objects):负责验证输入数据(如参数缺失、格式错误)。
- 响应对象(Response Objects):负责携带执行结果或错误信息,并能告知外部调用者操作是否成功。
为了让用例保持简洁,输入验证被移动到了请求对象中。
-
验证逻辑:本章以“房间列表”用例为例,增加了对过滤器(filters)的支持。请求对象会检查过滤器参数是否合法(例如:是否支持
price__gt等)。 -
工厂模式:引入了工厂函数
build_room_list_request。 -
如果参数合法,返回
RoomListValidRequest。 -
如果非法,返回
RoomListInvalidRequest,并包含具体的错误信息。 -
好处:用例无需再关心输入数据是否“长得正确”,只需处理业务逻辑。
无论执行成功与否,用例都应返回一个统一格式的响应对象。
- ResponseSuccess:包含成功后的数据(如房间列表)。
- ResponseFailure:包含错误类型(Type)和错误消息(Message)。
- 错误类型分类:定义了常见的错误类型,如
RESOURCE_ERROR(资源错误)、PARAMETERS_ERROR(参数错误)和SYSTEM_ERROR(系统/业务逻辑错误)。
在引入请求/响应模式后,用例的逻辑流程变为:
- 接收请求对象。
- 检查请求是否有效(
if not request:)。 - 如果请求无效,直接返回一个包含错误信息的
ResponseFailure。 - 如果请求有效,调用仓库层执行业务逻辑。
- 返回包含结果的
ResponseSuccess。
作者展示了这种模式如何让外部层(如 Web 框架或命令行工具)受益:
- Flask 适配:在控制器中,根据响应对象的
type映射到相应的 HTTP 状态码(如PARAMETERS_ERROR映射为 400)。 - CLI 适配:命令行工具只需根据响应对象的布尔值决定是打印结果还是打印错误消息。
- 鲁棒性:通过请求验证确保无效数据无法进入业务核心。
- 一致性:所有的用例都遵循相同的输入输出模式,使得前端/调用方能够预测错误格式。
- 独立性:业务逻辑与具体的传输协议(HTTP/CLI)完全解耦,错误处理逻辑是纯 Python 实现,易于测试。
本章的核心目的是演示如何将项目之前使用的内存存储(In-memory Repository)替换为真实的数据库(PostgreSQL),以此来展示整洁架构最大的优势之一:可以非常简单地替换现有组件(甚至底层技术完全不同),而不影响核心业务逻辑。
- 在之前的章节中,业务用例(Use Case)只通过存储库对象暴露的 API(例如
list方法)与其进行交互,形成了一种非常松散的耦合。 - 只要新的存储库实现了相同的接口(多态),业务逻辑就不需要关心底层使用的是内存字典还是真实的 SQL 数据库。
- 同时,不同的存储库实现可以拥有不同的初始化方法(
__init__)。比如内存存储需要传入测试数据,而真实的数据库存储则需要传入数据库地址和凭证。
- 作者选择使用 PostgreSQL 作为真实的外部数据库,并引入 SQLAlchemy(一种流行的 Python ORM 框架)来进行数据库操作。
- 不要去 Mock(模拟)ORM:作者特别强调,试图去 Mock SQLAlchemy 的查询结构是非常糟糕的做法。这会导致测试代码极其复杂、难以编写且几乎无法维护。
- 正确的做法是进行集成测试(Integration Tests):连接真实的数据库,建立 SQLAlchemy 会话,执行测试后销毁数据库。
- 由于集成测试涉及真实的数据库读写和容器的创建/销毁,运行速度会比较慢。因此,最好不要每次运行测试套件时都执行它们。
- 使用 pytest 打标签:作者演示了如何通过
pytest.mark.integration为测试模块打上集成测试标签,并在pytest.ini中进行注册。 - 跳过默认执行:通过修改
tests/conftest.py,添加了一个自定义的命令行参数--integration。测试配置被修改为:默认情况下跳过所有集成测试,只有当开发者在运行 pytest 时显式加上--integration参数时,这些测试才会执行。
- 文章展示了如何使用 SQLAlchemy 的声明式基类(
declarative_base)来创建数据库表映射(例如Room类)。 - 核心理念区分:作者在此指出了一个整洁架构中的关键点——这里的 SQLAlchemy
Room类代表的是数据库表结构(存储层),而不是业务逻辑中的领域模型(Domain Model)。 虽然在这个简单的例子中两者字段相似,但在实际项目中,为了扩展性,存储结构和领域模型往往会分离。开发者需要自己负责在这两层之间进行数据映射和同步。
- 为了让集成测试顺利运行,后台必须有一个运行中的 PostgreSQL 实例。
- 作者推荐使用 Docker 来进行系统编排。通过编写管理脚本(或使用 Docker Compose),可以在测试开始前自动拉起一个干净的数据库容器,在测试完成后再将其销毁,从而保证测试环境的隔离性和一致性。
总结而言,本章通过引入 Postgres 数据库和 SQLAlchemy,展示了如何在整洁架构中平滑地接入真实的外部系统。并重点讲解了由此带来的工程问题:如何处理ORM测试(使用集成测试而非Mock)、如何管理缓慢的集成测试(Pytest标签过滤),以及如何管理测试依赖的环境(Docker编排)。
本章的核心主题是展示清洁架构的灵活性:通过将数据库从上一章的 PostgreSQL(关系型数据库)切换到 MongoDB(非关系型数据库/NoSQL),证明在不改变核心业务逻辑的情况下,替换外部系统是多么简单。
作者希望通过实现一个新的 MongoRepo 类,向读者展示清洁架构如何实现“存储系统无关性”。由于之前的章节已经搭建好了测试框架,本章重点在于展示与 MongoDB 相关的特定实现。
作者利用了第 6 章建立的测试结构,在 tests/repository/mongodb/conftest.py 中定义了针对 MongoDB 的 pytest 固件:
mg_database_empty: 负责创建 MongoDB 客户端并初始化/清理测试数据库。mg_test_data: 提供与 Postgres 测试相同的数据集,确保测试的一致性。mg_database: 向数据库注入初始数据。- 依赖管理: 在生产环境依赖(
requirements/prod.txt)中增加了pymongo。
为了支持 MongoDB 的运行和测试,需要更新环境配置:
- Docker Compose: 在
docker/testing.yml中增加了一个临时 MongoDB 容器镜像,并指定了非标准端口(27018)以避免与本地实例冲突。 - 配置文件: 更新
config/testing.json,添加 MongoDB 的主机名、端口、用户名和密码等连接信息。
作者强调,由于 MongoDB 支撑的是相同的用例(Use Case),其集成测试逻辑(test_mongorepo.py)几乎是 PostgreSQL 测试的镜像:
- 测试不带参数的列表查询。
- 测试带有各种滤镜(如
code__eq,price__lt,price__gt等)的查询。 - 特别测试: 增加了一个将价格作为字符串传入的测试(
test_repository_list_with_price_as_string),以确保实现中处理了类型转换(MongoDB 对类型敏感)。
这是本章的技术核心,展示了如何在不影响 Domain 层(领域层)的情况下实现 MongoRepo:
- 初始化: 使用
pymongo建立连接。 _create_room_objects: 私有方法,将 MongoDB 返回的原始字典数据转换为领域模型room.Room对象。list方法:- 将清洁架构通用的滤镜格式(如
price__lt)映射为 MongoDB 特有的查询语法(如{"price": {"$lt": value}})。 - 这种映射逻辑确保了 Repository 接口对 Usecase 层保持透明。
通过这一章的实践,作者总结了两个关键点:
- 层级隔离的价值: Usecase 和 Domain 层完全感知不到数据库从 SQL 变成了 NoSQL,代码无需做任何修改。
- 测试架构的威力: 良好的测试固件设计使得接入新存储系统的开发工作变得非常高效。
作者指出,“生产就绪”意味着需要超越开发用的 Flask 服务器,构建一个完整的技术栈。该架构包含三个核心组件:
- Web 服务器 (Nginx): 作为反向代理和负载均衡器,负责接收外部 HTTP 请求。
- WSGI 服务器 (Gunicorn): 运行 Flask 应用,负责处理并发请求(示例中配置了 4 个工作进程)。
- 数据库 (PostgreSQL): 真正的外部持久化存储,而非之前章节中使用的内存存储。
为了模拟真实生产环境,本章详细讲解了如何使用 Docker Compose 组织服务:
- 配置文件: 创建了
config/production.json来定义生产环境的环境变量(如FLASK_DEBUG=0)。 - Docker Compose: 编写了
docker/production.yml,定义了三个容器:db(Postgres)、web(Flask + Gunicorn) 和nginx。 - 持久化: 在 Docker 配置中使用了
volumes,确保数据库数据在容器重启或关闭后不会丢失。
作者扩展了项目的 manage.py 脚本,使其能够加载 JSON 配置并自动执行 Docker 命令。
- 新增了
compose命令,允许开发者通过./manage.py compose up -d这样的指令直接启动整套生产栈。 - 通过脚本自动处理环境变量注入,避免了手动配置的繁琐。
这是展示“整洁架构”威力的关键点。得益于之前的接口设计(Repository Pattern),将应用从 MemRepo(内存仓库)切换到 PostgresRepo(Postgres 仓库)变得极其简单:
- 最小代码改动: 只需在 Flask 的蓝图(Blueprint)中更改一行代码,将初始化仓库的类从内存版改为 Postgres 版。
- 无感切换: 业务逻辑(Use Cases)完全不需要修改,因为它们只依赖于抽象接口,而不关心数据是存在内存里还是数据库里。
为了处理数据库模式(Schema)的管理,本章引入了 Alembic(SQLAlchemy 的迁移工具):
- 配置 Alembic: 讲解了如何让 Alembic 能够读取应用的环境变量。
- 自动生成迁移: 使用
alembic revision --autogenerate根据代码中的模型自动创建数据库表。 - 应用迁移: 通过
alembic upgrade head执行迁移,确保数据库结构与代码同步。
本章最后对全书的核心实践进行了总结:
- 解耦: 通过解耦,系统可以轻松支持不同的存储系统(Postgres, MongoDB, 内存)。
- 健壮性: 系统拥有完善的错误管理和请求/响应处理机制。
- 测试驱动: 架构的设计使得测试(单元测试和集成测试)变得非常容易,从而保证了代码质量。
核心意义: 这一章证明了整洁架构不仅是理论上的“优雅”,在实际部署和运维中也具有极高的灵活性,能够快速适应从简单原型到复杂生产环境的转变。