qwil的六边形结构框架
gasofo的Python项目详细描述
gasofo(qwil的六边形代码体系结构框架)
gasofo是qwil采用的实现六边形代码体系结构 (1, 2, 3, 4)。
有关gasofo如何用于构建应用程序的示例,请参见/example
。
安装Gasofo
pip安装gasofo
定义服务
服务应该是无状态的,并且应该只通过其"需求"端口访问其他服务的资源。功能 由服务提供的通过"提供"端口公开。
服务可以这样定义:
fromgasofoimportService,Needs,providesclassMyService(Service):deps=Needs(['some_data','another_service'])# ports this service 'Needs'@providesdefmy_feature(self,x):data=self.deps.some_data()# needs are accessed via self.deps.<port_name>more_data=self.deps.another_service(value=x)returndata+more_data
在这里,我们用几个需求端口和一个名为 "My_功能"。
该类实例上的方法可以像普通类一样调用,但只能调用标记为
与@一起提供的
可以作为端口被发现。
服务的依赖项是通过needs port通过self.deps.port_name(…)
访问的。这些港口将
在应用程序连接时注入实际的提供程序函数。在端口连接到
提供程序将引发一个断开连接的端口
异常。
通过在服务类或实例上调用get_needs()
和get_provides()
可以查询服务的端口。
这有助于可视化和自动连接服务,形成业务域和完整的应用程序。
声明的验证
如果出现以下情况,则在类构造期间(通常是在导入模块时)将引发异常 验证失败:
- 不允许构造函数(
\uu init
),因为服务是无状态的。 - 所有
self.deps.<;port_name>;
都必须引用已声明的需要端口。 - 类中的任何方法都必须至少引用一次所有声明的需要端口。
- 所有端口名必须以小写字母开头,并且只能包含字母数字字符或下划线。
- 端口名不能与保留名称之一匹配,例如
get_needs
,get_provides
等。有关完整列表,请参阅gasofo.ports.reserved_port_names
将需要端口声明为接口
以上各节中的示例声明需要使用端口名列表的端口。这非常方便快捷, 但不太适合IDE或测试。
推荐的方法是使用needsInterface
声明needs端口
fromgasofoimportService,NeedsInterface,providesclassMyServiceNeeds(NeedsInterface):defsome_data(self):# type: () -> int"""A brief description."""defanother_service(self,value):# type: (int) -> int"""You can include as much doc here as you like."""classMyService(Service):deps=MyServiceNeeds()@providesdefmy_feature(self,x):data=self.deps.some_data()# needs are accessed via self.deps.<port_name>more_data=self.deps.another_service(value=x)returndata+more_data
使用 就gasofo而言,类型提示是可选的,但我们鼓励使用它。这些端口只连接到
运行时的具体实现,因此类型提示是IDE推断
参数和返回值。额外的努力是值得的! PyCharm中的代码导航说明:通常的"查找用法"和"转到声明"功能将照常工作
但这只允许您在needsInterface类中的deps用法和stubs方法之间跳转。这个
提供程序实现为n不是静态地联系在一起,因此不能被妖精发现。找到一个
匹配的提供程序端口将使用"转到符号"功能(导航>;符号)将查找
符号。我们建议为此创建一个自定义的键映射快捷方式——我使用super+鼠标右键单击,它允许我
快速单击任何deps或needs存根并找到其他定义。' 当我们使用 已发布的端口名与实际方法名之间的不匹配可能会导致混淆,因此请谨慎使用。 我们目前不使用这些标志,因此暂时不使用文档:) 域是组件(服务或其他域)的集合,这些组件被组合在一起并封装成更高的
级别业务组件。包含组件的端口的子集将发布为
域和组件的所有需要端口(这些端口在内部无法通过匹配提供来实现)都公开为
域的需要。 域类不应包含任何其他属性、方法或构造函数。 与服务一样,可以通过在域类上调用 在实例化时,代理方法被动态绑定到域对象,这样提供的端口也可以被访问
作为方法调用,即 对于具有大量内部组件和大量预期提供端口的域,手动定义它们并保持
它们是最新的,可能是一件麻烦事。 对于这种情况,请使用 在最简单的用例中,可以通过调用
因此,如果端口是在更高级别(即组件级别)完成的,则建议布线。为了
例子: 这里的先决条件是提供者的端口名必须与使用者的端口名匹配。我们相信这是
一件好事——在应用程序中使用全局唯一的端口名来表示意图和兼容性
容易推理AB输出端口并允许自动布线。 上面提到,在实例化时,域将自动实例化所有底层服务,并且
根据端口名自动连接它们。您可以使用 如上所述,建议的布线方法是在组件级别进行布线。这意味着任何
我们希望在布线中包含实现gasofo.iprovideneeds接口的好处
接口的需要。使用
@为u提供
@provides
定义provides端口时,端口的名称将取自方法名。在某些情况下
如果我们希望端口名与实际的方法名不同,可以使用@为u提供
fromgasofoimportService,provides_withclassMyService(Service):@provides_with('db_get_blah')defget_blah(self,blah_id):# ...
@为u提供
还允许我们附加附加的元数据(标志)端口,例如
@为您提供('db_get_blah',web_only=true)
定义域
fromgasofoimportDomainfrommyproject.servicesimportMyService,AnotherServiceclassMyDomain(Domain):__services__=[MyService,AnotherService]# Components contained in this domain__provides__=['get_blah','do_something_else']# subset of ports from services defined in __services
服务应定义为组件(服务或域)类的列表,而不是实例。一个例子
当域被实例化时,这些组件中的每一个都将被实例化,并且
匹配的名称将自动连接在一起。
get_needs()
和get_provides()
来查询域的端口
或实例。我的域实例。我的端口(…)
。这很方便,但目前不是很方便
IDE友好——IDE不知道动态添加的方法和端口的底层argspec,所以代码
建议和类型检查不起作用。(如果我们发现自己需要
定期访问这些方法。)自动注册为域提供端口
autoprovide
:fromgasofoimportDomain,AutoProvidefrommyproject.servicesimportMyService,AnotherServiceclassMyDomain(Domain):__services__=[MyService,AnotherService]# Components contained in this domain__provides__=AutoProvide(pattern='db_.*')# auto export all ports that start with db_
autoprovide
允许以方便的方式发布所有提供的端口,这些端口与给定的regex模式匹配。如果一种模式
未提供,所有提供内部服务的端口都已公开。请尽量少用这个
通过查询mydomain.get\u provides()
连接应用程序
服务实例.deps.connect_端口(端口名'blah',func=some_callable)
。注意,这是对
service.deps
可连接到任何可调用的对象。一旦我们拥有超过
应用程序中的几个端口。c1=MyComponent()# This could be a Service or Domain c2=MyProvider()# Anything that implements IProvide, e.g. Service, Domain, or some custom implementationc1.set_provider(port_name='blah',provider=c2)# c1.deps.blah ---> c2.blah
自动布线
gasofo.auto_wire()
对实例化的组件执行相同的操作
你自己用。这通常是连接完整应用程序的方式。fromgasofoimportauto_wire,Domainfrommyapp.domainsimport*frommyapp.adaptersimport*classMyAwesomeApp(Domain):# encapsulate all my app domain into a single domain__services__=[DomainA,DomainB,DomainC,DomainD]__provides__=LIST_OF_PORTS_TO_EXPOSE_AT_APP_LEVELdefget_app():app=MyAwesomeApp()dependencies=[my_db_provider(),redis_provider(),logging_provider(),]auto_wire([app]+dependencies,expect_all_ports_connected=True)# raise if there are unfulfilled portsreturnapp
创建提供者的便利功能
这并不难做到,但涉及到不必要的锅炉板,以将它们包装在兼容的类结构中。
对于这种情况,您可以使用object_as_provider
或func_as_provider
自动包装对象或函数
在公开iprovide
接口的包装器中。
一些示例:
fromgasofoimportfunc_as_providerimporthashlib# creates provider which provides "get_md5_hash"md5_provider=func_as_provider(func=hashlib.md5,port='get_md5_hash')
fromgasofoimportobject_as_providerclassMyStack(object):def__init__(self):self.stack=[]defpush_to_stack(self,value):self.stack.append(value)defpop_from_stack(self):returnself.stack.pop()stack_provider=object_as_provider(provider=MyStack(),ports=['push_to_stack','pop_from_stack'])
fromgasofoimportService,Needs,providesclassMyService(Service):deps=Needs(['some_data','another_service'])# ports this service 'Needs'@providesdefmy_feature(self,x):data=self.deps.some_data()# needs are accessed via self.deps.<port_name>more_data=self.deps.another_service(value=x)returndata+more_data0
适配器
适配器允许我们在端口和依赖项的提供者之间注入逻辑。一种看法是 服务应该关注业务逻辑并访问端口以获取数据或执行某些操作。它不应该 关注如何提供依赖关系,或者结构在源位置是什么,而不是让它自己 用于处理更多机械操作的适配器,如传输、序列化/反序列化、有效负载转换等。
例如,提供特定数据集的服务,以及需要该数据集但在 不同的格式。我们不需要为不同的格式提供多个提供程序端口,而是可以让所有的使用者 连接到同一个提供商,但每个提供商都有不同的适配器来处理重新格式化。
另一个例子是当将服务移动到另一个进程时,我们可以简单地引入适配器 rest或grpc调用连接,这些连接现在跨越进程,而服务本身没有任何更改。
在qwil中,我们使用两种适配器:
- 基于服务的适配器
- 注入适配器 < > >
- 域可视化(无需实例化服务/域) <Li>应用可视化(域/服务被实例化并连接起来)
- 实时序列图
基于服务的适配器
基于服务的适配器本质上是标准提供程序,即公开ined和iprovide接口的对象。他们 技术上与服务没有区别,只是它们不包含业务登录,而是将服务器作为一个桥 两个端口之间。
通过确保在整个应用程序中使用全局唯一的端口名,并确保 名称是兼容的,我们可以简单地抛出具有相应名称的基于服务的适配器来处理 不兼容并让自动布线过程将其连接起来。
例如,假设服务A提供者端口X和该数据是服务B和C所需要的。 数据格式略有不同。而不是c声明需要x,然后用数据污染它的业务逻辑 转换时,它应该用不同的端口名声明需要,并依赖适配器进行重新格式化。
fromgasofoimportService,Needs,providesclassMyService(Service):deps=Needs(['some_data','another_service'])# ports this service 'Needs'@providesdefmy_feature(self,x):data=self.deps.some_data()# needs are accessed via self.deps.<port_name>more_data=self.deps.another_service(value=x)returndata+more_data1
注入适配器
(尚未实施)
注入的适配器是通过在连接端口时注入的可调用函数调用的。这就行了 布线时。
注入可以是针对性的(即在特定端口的连接之间注入)或应用程序范围的(全部注入 连接)。后者主要用于调试/开发场景,用于检测端口调用,例如用于实时 序列图、性能分析、详细日志记录。
可视化
可视化非常重要,因为它将允许我们对应用程序和更高层次的抽象进行推理,并且 以直观的方式确认组件确实按照我们预期的方式连接。
(尚未实施)
测试
正确完成后,使用gasofo编写的应用程序和组件非常适合 排列动作断言"/ 指定的时间 测试-由于组件是无状态的,"givens"可以通过简单地设置需求端口和 "whens"是对提供端口的调用。
只要所有依赖项都正确声明为端口而不是 直接从服务中访问。
有关如何测试用gasofo编写的组件的一些示例,请参见/tests/example/
。
基本知识
对于每个测试场景,我们应该只附加被测行为显式需要的端口。所有 其他端口应保持未连接状态,以确保在访问意外依赖项时测试将失败。
在测试中附加端口可以手动完成,即准备提供程序并将其分配给服务端口。为了 例如,假设我们有一个时钟服务定义为:
fromgasofoimportService,Needs,providesclassMyService(Service):deps=Needs(['some_data','another_service'])# ports this service 'Needs'@providesdefmy_feature(self,x):data=self.deps.some_data()# needs are accessed via self.deps.<port_name>more_data=self.deps.another_service(value=x)returndata+more_data2
我们可以这样测试:
fromgasofoimportService,Needs,providesclassMyService(Service):deps=Needs(['some_data','another_service'])# ports this service 'Needs'@providesdefmy_feature(self,x):data=self.deps.some_data()# needs are accessed via self.deps.<port_name>more_data=self.deps.another_service(value=x)returndata+more_data3
这将工作,是相当干净的,但需要相当多的样板代码。我们可以进一步简化 通过使用gasofo.testing.attach_mock_provider
gasofo.testing.attach_mock_provider
这是生成能够满足服务的一个或多个端口的提供程序的简便方法。使用这个助手, 上面的测试可以重写为:
fromgasofoimportService,Needs,providesclassMyService(Service):deps=Needs(['some_data','another_service'])# ports this service 'Needs'@providesdefmy_feature(self,x):data=self.deps.some_data()# needs are accessed via self.deps.<port_name>more_data=self.deps.another_service(value=x)returndata+more_data4
attach_mock_provider
生成提供ports
参数中定义的端口的provider对象,然后附加
此提供程序的使用组件。调用中未定义的使用者上的任何端口都将保留
未附加。
注意,上面的 使用 为了编写更简洁的测试,还可以使用 例如,要测试上面定义的 值得注意的是, 也要通过 有关更多示例,请参见attach_mock_provider还返回provider对象,其中所有生成的模拟端口都可以作为
这个物体。这些属性是
mock.mock
对象的实例,允许我们进行更详细的测试设置,例如
fromgasofoimportService,Needs,providesclassMyService(Service):deps=Needs(['some_data','another_service'])# ports this service 'Needs'@providesdefmy_feature(self,x):data=self.deps.some_data()# needs are accessed via self.deps.<port_name>more_data=self.deps.another_service(value=x)returndata+more_data
5
ports
参数声明为列表而不是dict。
默认情况下不设置模拟的返回值。attach_mock_provider的一个额外好处是,如果组件需要定义为
needs接口
实例,然后使用mock.create\autospec
创建端口的下划线模拟对象。这将表明
对它的所有调用都遵循需求端口的argspec,从而验证服务方法是否正在访问deps
如预期。(现在完成这张图片唯一缺少的是连接时间断言,连接需要和
提供端口具有兼容的argspec)给出时间
gasoftestcase
基类包装掉大多数测试设置
并提供将测试构造为一系列给定的"当-时"调用的能力。时钟
服务fromgasofoimportService,Needs,providesclassMyService(Service):deps=Needs(['some_data','another_service'])# ports this service 'Needs'@providesdefmy_feature(self,x):data=self.deps.some_data()# needs are accessed via self.deps.<port_name>more_data=self.deps.another_service(value=x)returndata+more_data
6
self.given
调用返回创建的模拟对象,而self.when
调用返回
端口调用的实际输出。self.given
和self来探索其他参数的支持,然后
因为它们提供了声明更多参数的方法
复杂的要求,例如为给予者设置副作用或指定我们不关心
预期产量。gasoftestcase
还提供断言方法来断言需求端口按预期调用。这可能是
简单的断言,或者更复杂的断言是
打电话。例如:fromgasofoimportService,Needs,providesclassMyService(Service):deps=Needs(['some_data','another_service'])# ports this service 'Needs'@providesdefmy_feature(self,x):data=self.deps.some_data()# needs are accessed via self.deps.<port_name>more_data=self.deps.another_service(value=x)returndata+more_data
7
tests/example/domains/coffee_orders/test_order_history_service.py
。两个测试类
此文件中定义的--OrderHistoryServiceTestSimplified
和OrderHistoryServiceTestWithOutFramework
--是
与之等价,但后者实现时不带gasoftestcase
更高级别的测试,即域、应用程序、集成、验收测试
为域编写测试与测试服务相同,因为它们都实现相同的接口。
应用程序级的测试以及集成/验收测试也可以用类似的形式表示,除了 测试的设置将更加详细。例如,可以将整个应用程序连接起来,而不使用边缘 依赖项,然后将整个网格视为单个域。然后我们可以使用上面描述的相同工具来 执行我们的验收测试或集成测试。
有关如何实现此目标的简单示例,请参见example/domains/test_a p p.py
。