はじめに
2021年11月20日から21日にかけて、JAWS Pankration 2021 ~Up till Down~が開催されました。筆者はその中で DDD on AWS Lambdaというタイトルのセッションを担当しました。このセッションでは、ヘキサゴナルアーキテクチャを利用してドメインモデルをAWS Lambdaファンクションに実装する方法、また制御の反転（IoC）を利用してユニットテストを容易にする方法について紹介しました。
この記事を書いたのは
今回、JAWS Pankrationは新たな試みとして、広く海外のコミュニティーメンバーにも参加してもらうために、全てのスライドを英語で記述しポケトークを利用して逐次通訳しながらセッションを行いました。海外のコミュニティーメンバーにも見てもらいたかったので、このために実装したサンプルアプリケーションのリポジトリのREADME.mdも英語で記述しましたが、JAWS Pankrationのセッションが終了した後、やはり日本語でもこの内容を紹介したいと思い、この記事で紹介することにしました。この記事を見て興味を持たれた方は、サンプルアプリケーションのGitHubのリポジトリをぜひご覧ください。
injectorを使わず、よりシンプルな実装に修正したコードを GitHub/aws-samplesの配下でサンプルプロジェクトとして公開しました。
こちらが新しいサンプルアプリケーションのGitHubのリポジトリです。
JAWS Pankration 2021に登壇した理由
マイクロサービスアーキテクチャが注目され、サービスの分割手法としてドメイン駆動設計への関心が再度高まると共に、自社のサービスにAWS Lambdaを利用した場合にどのようにドメインモデルを実装するか、またユニットテストを容易にするにはどうすれば良いかという質問をされる機会が増えました。そこで自分なりに考えた結果、ヘキサゴナルアーキテクチャの概念を利用してドメインモデルを外部から隔離し、また制御の反転（IoC）を利用することで、クラス間を疎結合にしてユニットテストを容易にするのが良いのではないかと仮説を立て、実際にサンプルアプリケーションを実装してみることにしました。実装してみると個人的には割と良いと感じたので、GitHubにサンプルアプリケーションとして公開し、JAWS Pankration 2021の場でこのアイデアを紹介してみようと思いました。JAWS PankrationにCFPを提出したところ、幸運にも採択されて発表することができました。
JAWS Pankration 2021で発表したトピック
JAWS pankration 2021のセッションでお話ししたトピックは以下の通りです。
ドメイン駆動設計
マイクロサービスアーキテクチャによるサービスの分割方式としてEric Evansのドメイン駆動設計が再び注目されています。ドメイン駆動設計はビジネスエキスパートが用いる用語をユビキタス言語として抽出し、ユビキタス言語が共通認識として利用される範囲をコンテキストの境界と位置付け、境界づけされたコンテキストでドメインモデルを定義します。ビジネス要件に対してドメインモデルを設計し、モデルに適切なデータと振る舞いを持たせることで、ドメインエキスパートの知識を正しく実装に反映させることを目的としています。多くの時間をかけてモデルの設計を行って満足し、その後の実装と乖離してしまうのではなく、アジャイル的にドメインモデル、ユニットテスト、実装を繋げて、繰り返し改善し開発していくスタイルが個人的には気に入っています。
ヘキサゴナルアーキテクチャ
ヘキサゴナルアーキテクチャはAlistair Cockburnが提唱したアーキテクチャパターンで、ポートとアダプターパターンとも呼ばれます。ヘキサゴナルアーキテクチャでは、アプリケーションを利用するドライバ（アクター）は特定のテクノロジー用のプロシージャ呼び出し、またはメッセージに変換される要求を送信し、それをアプリケーションポートに渡します。アプリケーションはドライバーのテクノロジーについては関知しません。アプリケーションからインフラストラクチャなど外部へ連携する場合は、ポートを介してアダプターに送信します。アダプターは連携先の受信テクノロジーに必要な形式に適切に変換します。
制御の反転（IoC）と依存性の注入（DI）
Robert C.Martinによって提唱された依存関係逆転の原則（DIP）は、上位レベルの実装コードは下位レベルの詳細の実装コードに依存してはいけないというものです。この原則に従ってクラス間の依存を抽象クラスと継承を用いて設計し、また疎結合を実現するために制御の反転（IoC）を行う依存性の注入（DI）を実現するために Python injectorライブラリを利用しました。 コンストラクタのパラメータに依存関係のインスタンスを渡すようにしました。
サンプルアプリケーションについて
サンプルアプリケーションはこちらのGitHubリポジトリで確認できます。このサンプルアプリケーションではワクチンの予約システムを例に一部の機能のみを実装しています。ワクチンの予約システムには様々なビジネス要件がありますが、ここでは以下の要件を定義、実装しています。
- ワクチン接種希望者は、自分のIDと空き時間スロットのIDを指定して予約を行うことができる
- ワクチン接種希望者は、一人につき2つまで空き時間スロットの予約を取ることができる
- ワクチン接種希望者は、同一日時のスロットを予約することはできない
これらのビジネス要件をドメインモデルで実装しました。
クラスダイアグラムとシーケンスダイアグラム
サンプルアプリケーションのクラスダイアグラムは以下の通りです。DIPの原則に従い、ポートクラスとアダプタークラスは抽象クラスを継承する形で実装しています。
またクラス間のシーケンスダイアグラムは以下の通りです。
クラスの実装
サンプルアプリケーションのクラスはクラスダイアグラムで示したようにドメインモデル、ポート、アダプターに分かれます。また各クラス間はインターフェイスの役割を持つ抽象クラスへ依存する形となっています。Pythonで実装しているため、インターフェイスの代わりに抽象クラスを利用しています。
ドメインモデル（抜粋）
ドメインモデルはシンプルな実装になっています。
from slot import Slot
'''
Recipient: Domain Model
'''
class Recipient:
def __init__(self, recipient_id:str, email:str, first_name:str, last_name:str, age:int):
self.__recipient_id = recipient_id
self.__email = email
self.__first_name = first_name
self.__last_name = last_name
self.__age = age
self.__slots = []
@property
def recipient_id(self):
return self.__recipient_id
@property
def email(self):
return self.__email
@property
def first_name(self):
return self.__first_name
@property
def last_name(self):
return self.__last_name
@property
def age(self):
return self.__age
@property
def slots(self):
return self.__slots
def are_slots_same_date(self, slot:Slot) -> bool:
for selfslot in self.__slots:
if selfslot.reservation_date == slot.reservation_date:
return True
return False
def is_slot_counts_equal_or_over_two(self) -> bool:
if len(self.__slots) >= 2:
return True
return False
def add_reserve_slot(self, slot:Slot) -> bool:
if self.are_slots_same_date(slot):
return False
if self.is_slot_counts_equal_or_over_two():
return False
self.__slots.append(slot)
slot.use_slot()
return True
インターフェイスとなる抽象クラス（一部）
インターフェイスとして抽象クラスを利用しています。ユニットテストでは、Amazon DynamoDBへアクセスする具象クラスの代わりに、この抽象クラスを継承するダミークラスを定義しテストを行っています。
以下はドメインモデルがポートクラスを経由してアクセスするアダプタークラスのインターフェイスの例です。
rom abc import ABCMeta, abstractmethod
from recipient import Recipient
'''
interface of Recipient adapter
'''
class IRecipientAdapter(metaclass=ABCMeta):
@abstractmethod
def load(self, recipient_id:str) -> Recipient:
raise NotImplementedError()
@abstractmethod
def save(self, recipient:Recipient) -> bool:
raise NotImplementedError()
インジェクトされるクラス
インジェクトされるクラスは抽象クラスにのみ依存しておりインジェクトされるクラスの実装とは疎結合になっています。この例ではアウトプット用ポートクラスのRecipientOutputPortクラスが外部へアクセスするためにアダプタークラスを利用していますが、上に挙げたRecipientAdapterの抽象クラスを参照しています。
from recipient import Recipient
from i_recipient_output_port import IRecipientOutputPort
from i_recipient_adapter import IRecipientAdapter
'''
implementation of Recipient output port
'''
class RecipientOutputPort(IRecipientOutputPort):
def __init__(self, adapter:IRecipientAdapter):
self.__adapter = adapter
def get_recipient_by_id(self, recipient_id:str) -> Recipient:
return self.__adapter.load(recipient_id)
def add_reservation(self, recipient:Recipient) -> bool:
return self.__adapter.save(recipient)
Lambdaファンクションのハンドラ
LambdaファンクションのハンドラではDynamoDBへのアクセスを行うコンクリートクラスをインスタンス化してインジェクトしています。app.pyはRecipientInputPortクラスのメソッドを経由してドメインモデルへアクセスします。
def get_recipient_input_port():
return RecipientInputPort(
RecipientOutputPort(DDBRecipientAdapter()),
SlotOutputPort(DDBSlotAdapter()))
def lambda_handler(event, context):
'''
API Gateway event adapter
retrieve reservation request parameters
ex. '{"recipient_id": "1", "slot_id":"1"}'
'''
body = json.loads(event['body'])
recipient_id = body['recipient_id']
slot_id = body['slot_id']
# get an input port instance
recipient_input_port = get_recipient_input_port()
status = recipient_input_port.make_reservation(recipient_id, slot_id)
return {
"statusCode": status.status_code,
"body": json.dumps({
"message": status.message
}),
}
ユニットテストコード（抜粋）
ユニットテストではダミークラスをインジェクトしてテストを実施しています。ここでは、DynamoDBへアクセスするアダプタークラスのダミークラスを差し込んでテストを実施しています。
# value for testing
slot_id = "1"
reservation_date = datetime(2021,12,20, 9, 0, 0)
location = "Tokyo"
class DummySlotAdapter(ISlotAdapter):
def load(self, slot_id:str) -> Slot:
return Slot(slot_id, reservation_date, location)
@pytest.fixture()
def fixture_slot_output_port():
#SetUp
slot_output_port = SlotOutputPort(DummySlotAdapter())
# execute testing
yield slot_output_port
#TearDown
slot_output_port = None
def test_slot_output_port_slot_by_id(fixture_slot_output_port):
target = fixture_slot_output_port
slot_id = "1"
slot = target.get_slot_by_id(slot_id)
assert slot != None
assert reservation_date == slot.reservation_date
assert location == slot.location
サンプルアプリケーションのビルドとデプロイ
GitHubリポジトリからプロジェクトを取得したら以下の手順でビルドとデプロイが可能です。
Serverless Application Model
サンプルプログラムはAWS Serverless Application Model (SAM) でビルド、デプロイするようになっています。このプロジェクトには以下の内容が含まれています。
- src - アプリケーションのLambdaファンクション
- events - ファンクションのローカル実行で利用できるイベント
- tests/unit - アプリケーションコードのユニットテスト
- template.yaml - アプリケーションのAWSリソースが定義されたSAMテンプレート
ビルドとデプロイ
サンプルアプリケーションをビルド、デプロイするにはSAM CLIを利用します。事前にSAM CLIをインストールしておく必要があります。SAM CLIのインストールの詳細はこちらをご覧ください。
ビルドでデプロイを実行するためには以下のコマンドを実行します。
sam build --use-container
sam deploy --guided
初期データの登録
初期データを登録するAWS CLIを利用したコマンドが用意されています。事前にAWS CLIのインストール、アカウント情報の設定が必要です。以下のshellコマンドを実行します。
chmod +x setup/add_ddb_data.sh
setup/add_ddb_data.sh
SAMビルドとローカルでの実行
Lambdaファンクションをローカルで実行するには以下のコマンドを実行します。
sam build --use-container
sam local invoke HelloWorldFunction --event events/event.json
ユニットテストの実行
このサンプルアプリケーションは、Lambdaファンクションのユニットテストのサンプルにもなっています。ユニットテストを実行するためには、事前にユニットテストに必要なモジュールをインストールします。
pip install -r tests/requirements.txt --user
ユニットテストを実行します。
python -m pytest tests/unit -v
まとめ
ヘキサゴナルアーキテクチャのコンセプトを利用してAWS Lambdaファンクションでドメインモデルを実装してみました。この記事が皆さんにとって少しでも参考になれば幸いです。
本投稿は、個人の意見で、所属する企業や団体は関係ありません。
また掲載しているサンプルプログラム等の動作に関しても一切保証しておりません。
サンプルプログラムの変更に伴って一部の記事を更新しました。