序言 或许是为了显摆，也或许是虚心学习，总之我在去年年初花了大约两个月读完了《架构整洁之道》 。但读过后也仅仅就是读了而已，尽管书中描绘了一个名为整洁架构的软件架构，但我并没有理解并应用到实际的开发中去。书中的诸多理念最终都蛰伏在了我的脑海深处。
今年年初的时候我换了工作。新的单位给每人都配备了办公用的电脑，从此我也不用背着2公斤重的MacBook Pro通勤了。美中不足的地方是，我和cuckoo 之间的联系被斩断了，因为cuckoo是个单机程序，要在私人电脑和办公电脑上各装一份太不方便了。于是乎，我决定开两个新的项目，将cuckoo拆分为客户端和服务端两部分。
正好，这给了我在实际的项目中践行整洁架构的机会。
什么是整洁架构 不像数学领域的概念往往有一个精确的定义，书中甚至没有道出整洁架构是什么。相对的，只有一副引人入胜的架构示意图（图片摘自作者博客的这篇文章 ）
在作者的文章 中，对图中的四个层次给出了响应的解释：
Entities封装了企业范围内的业务规则。如果你没有经营一个企业，仅仅是开发一款应用，那么Entities就是应用的业务对象，它们封装了应用内最通用、上层的规则。
Use Cases包含了与应用相关的业务规则。它封装并实现了系统的所有用例。
这一层负责将最方便entities和use cases的数据转换为最方便外部系统使用的格式。在这一层以内都是抽象的，对外界诸如MVC、GUI、数据库等均是无感知的。此外，这一层也负责与外部服务通信。
Frameworks & Drivers，顾名思义，这一层包含了与框架相关的代码，或者像C语言中的main函数这样的入口函数代码；
如何应用整洁架构
实际项目的例子 前文提到，为了满足新需求，我需要将cuckoo 改造为C/S模型。但比起缓缓地将cuckoo拆解为两部分，我更乐于大刀阔斧地从头开发开发这两个程序，于是便诞生了：
服务端程序为nest ，负责管理任务、计划等实体对象，并提供基于HTTP协议的API；
客户端程序为fledgling ，负责与nest通信，并在客户机上触发通知（如macOS的右上角弹出通知）。
它们都是我依照自己对整洁架构的理解来编写的。
从架构理念到具体决策 正如REST 仅仅是一种软件结构风格而不是具体的设计指南一样，整洁架构也并没有规定示意图中的分层结构该如何运用一门语言的特性来实现，这需要开发者自己去摸索。下文我给出自己在nest和fledgling项目中的做法。
如何安排代码目录结构 在程序的代码结构中，最接近于架构示意图的分层架构的，当属代码仓库的目录结构了。模仿整洁架构中的四层结构，我在nest中也安排了相似的目录结构
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 (venv) ➜ nest git:(master) tree -I '__pycache__' -d ./nest ./nest ├── app │ ├── entity │ └── use_case ├── cli │ ├── command │ └── config ├── infra ├── repository │ └── DDL └── web ├── config ├── controller └── presenter 13 directories
nest/app/entity/目录nest/app/entity/目录下的各个文件分别定义了系统中的各个实体类型
1 2 (venv) ➜ nest git:(master) ls nest/app/entity __pycache__ certificate.py location.py plan.py task.py user.py
例如：
task.py中定义了类Task，表示一个任务；plan.py中定义了类Plan，表示任务的一次触发计划，等等。
entity/目录下的各个文件中，还定义了管理各种实体对象生命期的仓库对象，例如：
task.py中定义了类ITaskRepository，它负责增（add方法）删（clear、remove方法）查（find、find_by_id方法）改（同样是add方法）任务对象；plan.py中定义了类IPlanRepository，同样能够增（add方法）删（clear、remove方法）查（find_as_queue、find_by_id、find_by_task_id方法）改（同样是add方法）计划对象，等等。
实体类型都是充血模型，它们实现了系统核心的业务规则，例如：
类Plan有方法is_repeated用于检查是否为重复性任务；
有方法is_visible用于检查该计划在当前时间是否可见；
有方法rebirth用于生成一个新的、下一次触发的计划，等等。
这个目录下的内容相当于整洁架构中的Entities层。
nest/app/use_case/目录nest/app/use_case/目录下的各个文件分别定义了系统所提供的功能
1 2 3 (venv) ➜ nest git:(master) ls nest/app/use_case __init__.py authenticate.py change_task.py create_plan.py delete_plan.py get_location.py get_task.py list_plan.py login.py registration.py __pycache__ change_plan.py create_location.py create_task.py delete_task.py get_plan.py list_location.py list_task.py pop_plan.py
例如：
authenticate.py定义了系统如何认证发送当前请求的用户；change_task.py定义了系统如何修改一个任务对象，等等。
每一个处于该目录下的文件，只会依赖nest/app/entity/中的代码，并且它们都是抽象的。例如，authenticate.py中的类AuthenticateUseCase的构造方法中，要求其：
参数certificate_repository必须是类ICertificateRepository或其子类的实例；
参数params必须是类IParams或其子类的实例。
然而ICertificateRepository和IParams其实都是抽象基类ABC的子类，并且它们都有被装饰器abstractmethod装饰的抽象方法，因此并不能直接实例化。
该目录相当于整洁架构中的Use Cases层。
其它目录 顾名思义，cli和web目录分别是与命令行程序、基于HTTP的API相关的代码，它们实现了处理来自命令行和HTTP协议的输入，以及打印到终端和返回HTTP响应的功能。repository目录下的各个文件实现了entity目录中各个抽象的仓库类的具体子类
1 2 (venv) ➜ nest git:(master) ls nest/repository DDL __init__.py __pycache__ certificate.py db_operation.py location.py plan.py task.py user.py
例如：
certificate.py中实现了entity/目录下的同名文件中的抽象类ICertificateRepository——一个基于内存的子类MemoryCertificateRepository，以及一个基于Redis的子类RedisCertificateRepository；location.py中实现了entity/目录下的同名文件中的抽象类ILocationRepository——基于MySQL的子类DatabaseLocationRepository，等等。
需要注意的是，除了app外的这些目录，并不能与整洁架构示意图中的外面两层严格对应起来。例如，尽管cli和web的名字一下子就让人认为它们处于Frameworks & Drivers层，但web/presenter/目录下的内容其实与框架并无联系。反倒是从命名上看处于Interface Adapters层的web/controller/目录，其中的代码依赖于Flask框架。
如何往Use Cases层传入数据 在鲍勃大叔的文章中，提到了关于如何在层之间传递数据的原则
Typically the data that crosses the boundaries is simple data structures. You can use basic structs or simple Data Transfer objects if you like. Or the data can simply be arguments in function calls. Or you can pack it into a hashmap, or construct it into an object.
在nest/app/use_case/目录下的所有用例采用的都是这里提到的construct it into an object的方式。以create_task.py为例：
1 2 3 4 5 6 7 8 9 10 11 12 class IParams (ABC ): @abstractmethod def get_brief (self ) -> str : pass @abstractmethod def get_keywords (self ) -> List [str ]: pass @abstractmethod def get_user_id (self ) -> int : pass
用内置模块abc中的抽象基类ABC、装饰器abstractmethod，以及类CreateTaskUseCase中的assert一起模拟类似Java中的interface的效果；
用方法而不是成员变量来获取不同的输入参数：
get_brief获取任务的简述；get_keywords获取关键字列表；get_user_id获取创建该任务的用户的ID。
聪明的盲生已经发现了华点：明明只需要在类CreateTaskUseCase的构造方法中定义brief、keywords，以及user_id三个参数即可，为什么要用方法这么麻烦呢？答案是因为方法更灵活。
当你采用构造方法参数的方案时，本质上是立了一个假设：
在所有惯性系中，物理定律有相同的表达形式先完成所有参数的获取；再执行用例中的业务逻辑。
如果是一个基于HTTP协议的API，那么这个假设是成立的——用户在客户端发送的HTTP请求到达服务端后，便无法再补充参数了。但有一种场景，用户能够在用例执行业务逻辑的过程中，持续地与应用交互，那便是命令行程序。
我在fledgling项目中给了一个用户在用例执行过程中，交互式地输入的例子。在文件fledgling/app/use_case/delete_task.py中，实现了删除指定任务的用例。它要求输入两个参数
1 2 3 4 5 6 7 8 9 class IParams (ABC ): @abstractmethod def get_confirmation (self ) -> bool : """获取用户是否要删除该任务的确认。""" pass @abstractmethod def get_task_id (self ) -> int : pass
在文件fledgling/cli/command/delete_task.py中实现了IParams类的命令行形态。当没有从命令行参数中获取到任务的ID时，便会使用第三方库PyInquirer询问用户输入任务ID，并进一步确认
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 class Params (IParams ): def __init__ (self, *, task_id: Optional [int ] ): self .task_id = task_id def get_confirmation (self ) -> bool : if self .task_id: return True questions = [ { 'message' : '确定删除该任务' , 'name' : 'confirmation' , 'type' : 'confirm' , } ] answers = prompt(questions) return answers['confirmation' ] def get_task_id (self ) -> int : if self .task_id: return self .task_id questions = [ { 'message' : '输入要删除的任务的ID' , 'name' : 'task_id' , 'type' : 'input' , } ] answers = prompt(questions) return answers['task_id' ]
而这一切煮不在乎DeleteTaskUseCase并不会感知到，它独立于用户界面。
在哪一层维护业务规则 在《架构整洁之道》第20章中，鲍勃大叔给出了业务规则的定义
Strictly speaking, business rules are rules or procedures that make or save
业务规则往往不是独立存在的，它们需要作用在一些数据上
Critical Business Rules usually require some data to work with. For example, our loan requires a loan balance, an interest rate, and a payment schedule.
而整洁架构中的实体就是包含了一部分业务规则及其操作的数据的对象。以nest中的计划实体为例，在类Plan中包含了几种业务规则——尽管这些规则不能为我赚钱或者省钱：
一个计划的持续时长（如果有的话）不会是负的秒数——由duration的setter保障；
周期性计划必须指定周期——由new方法维护；
一个计划是重复的，当且仅当它有指定重复类型——由is_repeated方法维护；
一个计划是可见的，当且仅当它：
要么没有指定可见的小时，要么当且时间处于指定的小时中，并且；
要么没有指定星期几可见，要么今天是指定的weekday——由is_visible方法维护。
但在整洁架构的示意图中，Use Cases层也是有维护规则的，它维护的是应用的业务规则（Application Business Rules）。与Entities层所维护的业务规则不同，Use Cases层的业务规则取决于应用提供的功能。例如，在nest项目修改一个计划的用例ChangePlanUseCase类的方法run中，会：
检查指定的计划是否存在——显然，实体没法检查自己是否存在；
检查计划是否能被修改；
检查新的地点的ID是否指向真实存在的地点对象——显然，Plan对象不会去检查Location存在与否；
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 class ChangePlanUseCase : def run (self ): params = self .params plan_id = params.get_plan_id() plan = self .plan_repository.find_by_id(plan_id) if plan is None : raise PlanNotFoundError(plan_id=plan_id) if not plan.is_changeable(): raise UnchangeableError() found, location_id = params.get_location_id() if found: if location_id: location = self .location_repository.get(id_=location_id) if not location: raise LocationNotFoundError(location_id=location_id) plan.location_id = location_id
聪明的你一定发现了：is_changeable为什么不作为Enterpries Business Rules，在Plan对象内自行检查呢？答案是因为这样写更简单。
试想一下，如果要让Plan自己禁止在is_changeable为False时被修改，那么必须：
先为所有可修改的属性设置setter；
在每一个setter中都调用is_changeable进行检查。
之所以要这么做，是因为一个实体对象（在这里是指Plan的实例对象）是外部的时间流动是无感知的。它不知道外层（此处是Use Cases层）会先 调用哪一个方法，后 调用哪一个方法。因此，要想保持“终止状态的计划不能修改”，就必须在每一处setter都检查。
与之相反，在用例中有编排，因此它可以感知时间的流动。用例可以让Plan的is_changeable方法在其它任何方法之前被调用，因此免除了繁琐地在每一个setter中检查is_changeable的必要。
如何获取Use Cases层的处理结果 正如往Use Cases层中输入参数可以采用：
直接在__init__中传入对应类型的参数，或；
在__init__中传入一个能根据方法提取参数的对象。
两种方案一样，获取Use Cases层的计算结果同样有两种方案：
获取run方法的返回值，捕捉它的异常，或；
在__init__中传入一个能够接受不同结果并处理的对象。
在nest这样的仅仅提供HTTP API的应用中，第1种方案便已经足够了。例如，在文件nest/web/controller/create_plan.py中，类CreatePlanUseCase的run方法的返回值为创建的计划对象，如果run调用成功，这个controller会借助于PlanPresenter，将计划对象转换为JSON对象格式的字符串，返回给调用方；如果调用失败，那么controller中也会捕捉异常（如InvalidRepeatTypeError）并以另一种格式返回给调用方。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 def create_plan (certificate_repository, repository_factory ): params = HTTPParams() use_case = CreatePlanUseCase( location_repository=repository_factory.location(), params=params, plan_repository=repository_factory.plan(), task_repository=repository_factory.task(), ) try : plan = use_case.run() presenter = PlanPresenter(plan=plan) return { 'error' : None , 'result' : presenter.format (), 'status' : 'success' , }, 201 except InvalidRepeatTypeError as e: return { 'error' : { 'message' : '不支持的重复类型：{}' .format (e.repeat_type), }, 'result' : None , 'status' : 'failure' , }, 422
如果想要更高的灵活性并且也有施展的空间，那么可以考虑第2种方案。例如fledgling项目中文件fledgling/app/use_case/list_plan.py中，就定义了一个接口IPresenter
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 class IPresenter (ABC ): @abstractmethod def on_find_location (self ): pass @abstractmethod def on_find_task (self ): pass @abstractmethod def on_invalid_location (self, *, error: InvalidLocationError ): pass @abstractmethod def show_plans (self, *, count: int , plans: List [Plan] ): pass
并且在用例的执行过程中，会多次向self.presenter传递数据
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 class ListPlanUseCase : def run (self ): location_id = None location_name = self .params.get_location_name() no_location = self .params.get_no_location() if not no_location and location_name is not None : locations = self .location_repository.find(name=location_name) if len (locations) == 0 : self .presenter.on_invalid_location(error=InvalidLocationError(name=location_name)) return location_id = locations[0 ].id page = self .params.get_page() per_page = self .params.get_per_page() criteria = { 'page' : page, 'per_page' : per_page, } if location_id is not None : criteria['location_id' ] = location_id plans, count = self .plan_repository.list (**criteria) location_ids = [plan.location_id for plan in plans] self .presenter.on_find_location() locations = self .location_repository.find( ids=location_ids, page=1 , per_page=len (location_ids), ) task_ids = [plan.task_id for plan in plans] self .presenter.on_find_task() tasks = self .task_repository.list ( page=1 , per_page=len (task_ids), task_ids=task_ids, ) for plan in plans: location_id = plan.location_id location = [location for location in locations if location.id == location_id][0 ] plan.location = location task_id = plan.task_id task = [task for task in tasks if task.id == task_id][0 ] plan.task = task self .presenter.show_plans( count=count, plans=plans, ) return
在构造方法中注入presenter的缺点在于用例的run方法中需要显式地return，否则用例会继续执行下去。
Python语言特性的运用 模拟接口——abstractmethodv.s.NotImplementedError 整洁架构的每一层都只会依赖于内层，而内层又对外层一无所知，负责解耦两者的便是编程语言的接口特性。但Python并不像Java那般有interface关键字，因此我利用它的其它一系列特性来模拟出接口：
用class代替interface，这些类继承自内置模块abc的抽象基类ABC；
除此之外，这些类中的方法还用同一模块中的abstractmethod装饰，使它们必须由该类的子类全部定义；
在使用这个接口的位置（例如Use Cases层）用断言assert约束输入参数的类型。
nest中的大部分需要接口的位置我都是用这种手法来做的，但这种方式会给编写单元测试用例带来一些不便：
因为代码中用assert来检查参数类型，导致传入的参数只能是这个接口或其子类的实例；
因为接口类继承自ABC，所以必须定义所有被abstractmethod装饰的方法，否则在实例化时就会抛出异常。
例如，在nest项目的文件tests/use_case/task/test_list.py中，作为白盒测试的人员，我确切地知道类ListTaskUseCase的run方法只会调用它的task_repository的find方法，但在类MockTaskRepository中依然不得不定义基类的每一个方法——尽管它们只有一行pass语句。
如果愿意放弃一点点的严谨性，那么可以弱化一下上面的接口方案：
不使用abstractmethod，而是在本应为抽象方法的方法中只留下一句raise NotImplementedError；
不使用assert检查类型，而是在参数中写上type hint。
有了第1点，那么在测试用例中就不需要为测试路径上不会调用的方法写多余的定义了。而有了第2点，也就不需要为测试路径上不会引用的属性创建对象了，大可直接传入一个None。选择哪一种都无妨，取决于开发者或团队的口味。
金坷垃整洁架构的好处都有啥在《架构整洁之道》的第20章，作者给出了整洁架构的五种优秀特性：
独立于框架。例如，我可以花不是很大的力气，将nest从Flask 迁移到Bottle 上，尽管并不会无缘无故或频繁地这么做；
容易测试。例如，在nest项目的目录tests/use_case下的测试用例不需要有任何外部系统的依赖就可以编写并运行；
独立于用户界面。例如，在nest项目中同一个用例RegistrationUseCase就有HTTP API和命令行两种用户界面：
在文件nest/web/controller/registration.py中是HTTP API形态；
在文件nest/cli/command/register.py中则是命令行形态。
独立于数据库。例如，就像更换Web框架一样，我也可以从MySQL迁移到PostgreSQL中，这对于Entities和Use Cases层的代码而言别无二致；
独立于外部系统。例如，在fledgling项目中，尽管也定义了一个接口ITaskRepository，但不同于nest中基于数据库的实现子类DatabaseTaskRepository，在fledgling中实现的是基于网络传输的类TaskRepository。但究竟是基于单机数据库，还是身处一个分布式系统（C/S模型）中，Entities和Use Cases层对此是无感知的。
甘瓜苦蒂——整洁架构的不足 渗入内层的I/O
