ポートとアダプタのアーキテクチャPython + Nameko Microexample

導入

ポートとアダプタ(または六角形)アーキテクチャは、2005年にAlistair Cockburnによって導入されたソフトウェア設計の概念です.その主な目的は、アプリケーションロジックと外部の依存関係のデータベース、ユーザインタフェース、HTTPリクエストを提供するフレームワークなどを明確にすることです.

アプリケーションコアは、外部サービスと依存関係については、既存のポートを使用して全体のビジネスプロセスのオーケストレーションを提供する必要があります.ポートには次の2種類があります.

プライマリポート、アプリケーションコアを外部に公開するエントリです.通常はプライマリアダプタ(例えばREST API、CLIなど)で呼ばれるファブリックです.

二次ポートは、アプリケーションのコアを外部の世界(例えばデータベース、メール送信者など)と通信することができます.これは、二次アダプタによって補完されるインターフェイスです.

実装

Nameko 簡単なHTTPリクエスト、RPCとAMQPプロトコルのメッセージングを提供するマイクロサービスを構築するためのPythonフレームワークです.
簡単な例では、システムにエントリを登録し、システムの他の部分(または他のマイクロサービス)で処理するイベントを発行するアプリケーションを実装します.
アプリケーションコアからドメインイベントを発行できるインターフェイスとして二次ポートを定義します.実装によっては、ローカルイベントバスまたはRabbitmqのような本物のイベントブローカーにメッセージを公開することができました(ただし、実装はアダプタのResponsibility).

from abc import ABC, abstractmethod

class EventPublisher(ABC):
    @abstractmethod
    def publish(self, events):
        pass

二次アダプタを実装するにはEventPublisher なめこのラッパを作るEventDispatcher クラス.

# domain/events.py
from datetime import datetime
from typing import Dict
from dataclasses import dataclass

@dataclass(frozen=True)
class DomainEvent:
    @property
    def as_dict(self) -> Dict:
        serialized = asdict(self)
        serialized["name"] = self.name
        return serialized

@dataclass(frozen=True)
class RegisteredEntryEvent(DomainEvent):
      email: str 
      name: str = 'RegisteredEntryEvent'
      timestamp: datetime = datetime.now()

# infrastructure/event_publisher.py
from typing import List
from nameko.events import EventDispatcher
from application.event_publisher import EventPublisher
from domain.event import DomainEvent

class NamekoEventPublisher(EventPublisher):
    def __init__(self, dispatcher: EventDispatcher):
        self._dispatcher = dispatcher

    def publish(self, events: List[DomainEvent]) -> None:
        for event in events:
            self._dispatcher(event.name, event.as_dict)

この例のプライマリポートは、登録ユースケースを定義する単純なアプリケーションサービスです.EventPublisher インターフェイス(実装されていません)は、サービスに注入されます.

# application/service.py
from application.event_publisher import EventPublisher
from domain.event import RegisteredEntryEvent

class RegistrationService:
    def __init__(self, event_publisher: EventPublisher):
        self._event_publisher = event_publisher

    def register_entry(self, email: str) -> None:
        event = RegisteredEntryEvent(email=email)
        self._event_publisher.publish(event)

Nameko HTTP Endpointは、我々のための主要なアダプターですRegistrationEntryService ポート.HTTPリクエストデータを処理し、サービスメソッドを実行します.

import json
from nameko.events import EventDispatcher
from nameko.web.handlers import http
from werkzeug.wrappers import Request, Response

class NamekoRegistrationService:
    name = "nameko_registration_service"
    dispatcher = EventDispatcher()

    @http("POST", "/register")
    def register_entry(self, request: Request) -> Response:
        request_params = json.loads(request.data)
        email = request_params["email"]
        service = RegistrationService(
       event_dispatcher=NamekoEventPublisher(self.dispatcher),
        )
        service.register_entry(email)
        return Response(f"Registered entry for {email=}")

ここで重要なことは、アプリケーションコアにインフラストラクチャとAPIに関する知識がないことです.データベースアクセスをしたいなら、リポジトリインターフェイスを定義して、それを我々のサービスに注入しました.または、通知を送信する必要がある場合、我々はおそらく、インフラストラクチャ層のSMSまたはメール送信アダプタによって実装できる通知送信インターフェイスを作成します.しかし、アプリケーションコアの内部では、インフラストラクチャの実装について何も知らない(アプリケーションロジックが簡単に単体テストを行うことができます).そして、このアーキテクチャを使用する主な利益です.

概要

PythonとNamekoを使用した六角形のアーキテクチャベースの概念のmicroexampleであった.あなたがそれをおもしろいと思うならば、私はあなたが類似したプロジェクト(また、パイソン、なめこと港とアダプターも)の拡大された実現(ここでは省略されたドメイン層を含む)のために私のGithubリポジトリを訪問することを勧めます:https://github.com/jorzel/opentable . 六角形のアーキテクチャについてのより理論的な背景のためにhere .