qwil的六边形结构框架

gasofo的Python项目详细描述


gasofo(qwil的六边形代码体系结构框架)

gasofo是qwil采用的实现六边形代码体系结构 (1234)。

有关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_needsget_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

使用needs接口的好处

  • self.deps的属性不再是动态注入的,这意味着ide中的自动完成和建议 现在可以工作了。
  • 方法构造允许类型提示和docstring。
  • 函数签名现在是显式的,测试框架将使用它来断言端口被调用 带有预期参数。

就gasofo而言,类型提示是可选的,但我们鼓励使用它。这些端口只连接到 运行时的具体实现,因此类型提示是IDE推断 参数和返回值。额外的努力是值得的!

PyCharm中的代码导航说明:通常的"查找用法"和"转到声明"功能将照常工作 但这只允许您在needsInterface类中的deps用法和stubs方法之间跳转。这个 提供程序实现为n不是静态地联系在一起,因此不能被妖精发现。找到一个 匹配的提供程序端口将使用"转到符号"功能(导航>;符号)将查找 符号。我们建议为此创建一个自定义的键映射快捷方式——我使用super+鼠标右键单击,它允许我 快速单击任何deps或needs存根并找到其他定义。'

使用@为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

这里的先决条件是提供者的端口名必须与使用者的端口名匹配。我们相信这是 一件好事——在应用程序中使用全局唯一的端口名来表示意图和兼容性 容易推理AB输出端口并允许自动布线。

自动布线

上面提到,在实例化时,域将自动实例化所有底层服务,并且 根据端口名自动连接它们。您可以使用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

创建提供者的便利功能

如上所述,建议的布线方法是在组件级别进行布线。这意味着任何 我们希望在布线中包含实现gasofo.iprovide接口的需要。

这并不难做到,但涉及到不必要的锅炉板,以将它们包装在兼容的类结构中。

对于这种情况,您可以使用object_as_providerfunc_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_data
0

适配器

适配器允许我们在端口和依赖项的提供者之间注入逻辑。一种看法是 服务应该关注业务逻辑并访问端口以获取数据或执行某些操作。它不应该 关注如何提供依赖关系,或者结构在源位置是什么,而不是让它自己 用于处理更多机械操作的适配器,如传输、序列化/反序列化、有效负载转换等。

例如,提供特定数据集的服务,以及需要该数据集但在 不同的格式。我们不需要为不同的格式提供多个提供程序端口,而是可以让所有的使用者 连接到同一个提供商,但每个提供商都有不同的适配器来处理重新格式化。

另一个例子是当将服务移动到另一个进程时,我们可以简单地引入适配器 rest或grpc调用连接,这些连接现在跨越进程,而服务本身没有任何更改。

在qwil中,我们使用两种适配器:

  1. 基于服务的适配器
  2. 注入适配器
  3. < > >

    基于服务的适配器

    基于服务的适配器本质上是标准提供程序,即公开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_data
    1

    注入适配器

    (尚未实施)

    注入的适配器是通过在连接端口时注入的可调用函数调用的。这就行了 布线时。

    注入可以是针对性的(即在特定端口的连接之间注入)或应用程序范围的(全部注入 连接)。后者主要用于调试/开发场景,用于检测端口调用,例如用于实时 序列图、性能分析、详细日志记录。

    可视化

    可视化非常重要,因为它将允许我们对应用程序和更高层次的抽象进行推理,并且 以直观的方式确认组件确实按照我们预期的方式连接。

    (尚未实施)

    • 域可视化(无需实例化服务/域)
    • <Li>应用可视化(域/服务被实例化并连接起来)
    • 实时序列图

    测试

    正确完成后,使用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_data
    2

    我们可以这样测试:

    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
    3

    这将工作,是相当干净的,但需要相当多的样板代码。我们可以进一步简化 通过使用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_data
    4

    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.givenself来探索其他参数的支持,然后因为它们提供了声明更多参数的方法 复杂的要求,例如为给予者设置副作用或指定我们不关心 预期产量。

    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。两个测试类 此文件中定义的--OrderHistoryServiceTestSimplifiedOrderHistoryServiceTestWithOutFramework--是 与之等价,但后者实现时不带gasoftestcase

    更高级别的测试,即域、应用程序、集成、验收测试

    为域编写测试与测试服务相同,因为它们都实现相同的接口。

    应用程序级的测试以及集成/验收测试也可以用类似的形式表示,除了 测试的设置将更加详细。例如,可以将整个应用程序连接起来,而不使用边缘 依赖项,然后将整个网格视为单个域。然后我们可以使用上面描述的相同工具来 执行我们的验收测试或集成测试。

    有关如何实现此目标的简单示例,请参见example/domains/test_a p p.py

    欢迎加入QQ群-->: 979659372 Python中文网_新手群

    推荐PyPI第三方库


热门话题
java如何将cassandra中的行数据转换为与列相关的嵌套json   java如何使用jcr XPath在jcr:content/@jcr:data中搜索?   java在使用openCV进行安卓开发时如何利用手机的广角镜头   java解析扩展了接口,结束了一个潜在的无限循环   位置服务的@Override方法中存在java Android应用程序错误   java本地线程的用途和需求是什么   具有左右子访问的java节点树遍历   java验证JsonWebToken签名   JUL日志处理程序中的java日志记录   嵌入式Java读取给定时间段的串行数据。   java有没有办法从多个URL获取多个图像?   java线程通过等待intent阻止自己发送intent   java Spring MVC解析多部分内容请求   java JPA/Hibernate静态元模型属性未填充NullPointerException   java格式错误的字符(需要引号,得到I)~正在处理   java为什么PrintWriter对象抛出FileNotFoundException?   java Neo4j未正确保存标签   java IE不加载图像