👻
ドメイン駆動設計でサーバレスアプリを組む
１．はじめに
AWS上でサーバレスアプリを作る際に、クラス設計やモジュールの切り方をどうしたものかたまに悩むことがあります。やり方は色々あると思いますが、今回はサーバレスアプリにドメイン駆動設計（DDD）のエッセンスを取り入れることでいい感じの構造を作れないか試してみました。
なお、今回はサーバレス用フレームワークのChalice（Python）を使用したアプリケーションに対してDDDを適用しています。とはいえChaliceはFlaskなんかのフレームワークと書きっぷりがよく似てますし、言語がPythonであれば概ね他にも応用が利くのでは、と考えています。
２．DDD適用対象のソースコード
以前のブログ記事でも紹介した、OpenSearch（Elasticsearch）向けの商品検索APIサンプルコードに対してDDDを適用したいと思います。実際のコードは以下の通りで、処理の流れは「リクエストボディからJSON形式で検索条件を受け取り、OpenSearch用のクエリを生成して実行後、結果を整形してJSONで返却する」といった感じになっています。全ての処理をapp.pyに押し込んでおり、いい悪いは置いておいて比較的シンプルです。
app.py
import json
import traceback
import boto3
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
from chalice import Chalice
from chalice import Response
app = Chalice(app_name='product_search_sample')
# CloudFrontの制約で、GETではリクエストボディが使えないためPOSTにする
@app.route('/productSearch', methods=['POST'], cors=True)
def index():
try:
# 検索条件取得
searchCondition = getSearchCondition()
# 検索クエリの組み立て
query = createQuery(searchCondition)
# OpenSearchへの接続と検索
searchResultsFromOs = executeQuery(query)
# 返却用JSONの生成
responseData = createResponseData(searchResultsFromOs)
# API応答値の返却
return Response(
body = json.dumps(responseData, ensure_ascii=False),
headers = {'Content-Type': 'application/json'},
status_code = 200
)
except Exception as e:
# スタックトレース出力とエラー応答
print(traceback.format_exc())
responseData = {"message" : "内部エラーが発生しました"}
return Response(
body = json.dumps(responseData, ensure_ascii=False),
headers = {'Content-Type': 'application/json'},
status_code = 500
)
def getSearchCondition():
"""
リクエストボディから検索条件を抽出する
※入力チェックなどは特に行ってないので、必要に応じて実装
"""
body = app.current_request.json_body
# 念のためリクエストボディの内容を組み換え
searchCondition = dict()
if body.get("brand"):
searchCondition["brand"] = body["brand"]
if body.get("gender"):
searchCondition["gender"] = body["gender"]
if body.get("category"):
searchCondition["category"] = body["category"]
if body.get("price"):
searchCondition["price"] = body["price"]
if body.get("color"):
searchCondition["color"] = body["color"]
if body.get("size"):
searchCondition["size"] = body["size"]
return searchCondition
def createQuery(searchCondition):
"""
OpenSearchに対して投げるクエリを生成する
"""
# ベースとなるクエリ
query = {
"from" : 0,
"size": 50,
"track_total_hits" : True,
"sort" : [
{"productName" : {"order" : "asc"}}
],
"query" : {
"bool" : {
"must" : []
}
}
}
# 検索条件が存在する場合、MUST句（AND）に検索条件を詰め込む
if searchCondition:
for key in searchCondition.keys():
searchParameKey = key
searchParamValue = searchCondition.get(key)
if key == "price":
# 検索条件がpriceの場合は数値での条件を指定
query["query"]["bool"]["must"].append(
{
"range" : {
searchParameKey : {
"gte" : searchParamValue[0],
"lt" : searchParamValue[1]
}
}
}
)
else:
# price以外は文字列検索
query["query"]["bool"]["must"].append(
{
"terms" : {searchParameKey : searchParamValue}
}
)
return query
def executeQuery(query):
"""
OpennSearchへ接続し、検索クエリを投げて結果を返す
"""
# 接続文字列
host = 'xxxxxx.ap-northeast-1.es.amazonaws.com'
port = 443
region = 'ap-northeast-1'
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
indexName = 'test-products'
try:
# ES接続とクエリ実行
osClient = OpenSearch(
hosts = [{'host':host, 'port': port}],
http_auth = awsauth,
use_ssl = True,
verify_certs = True,
connection_class = RequestsHttpConnection
)
searchResultsFromOs = osClient.search(index=indexName, body=query)
return searchResultsFromOs
except Exception as e:
# スタックトレース出力
print(traceback.format_exc())
raise e
def createResponseData(searchResultsFromOs):
"""
OpenSearchの検索結果を受け取り、APIの応答値を生成する
"""
# 最終的な応答値の雛形を定義
responsData = {
"status" : "success",
"totalCount" : 0,
"results" : [],
}
# 検索結果が存在しない場合は早々に処理終了
totalCount = searchResultsFromOs["hits"]["total"]["value"]
if totalCount == 0:
return responsData
# 応答値の生成
responsData["totalCount"] = totalCount
for result in searchResultsFromOs["hits"]["hits"]:
source = result["_source"]
responsData["results"].append(
{
"productCode" : source["productCode"],
"productName" : source["productName"],
"brand" : source["brand"],
"gender" : source["gender"],
"category" : source["category"],
"price" : source["price"],
"color" : source["color"],
"size" : source["size"]
}
)
return responsData
ディレクトリ構成はこんな感じで、ほぼChaliceのデフォルトのままです。変化点といえば、Lambdaに適用するIAMポリシーをcustom-policy_dev.jsonに外出ししたことくらいでしょうか。この辺りについては前回記事に詳しく書いていますのでご参照ください。
product_search_sample/
├── app.py
├── .chalice
│ ├── config.json
│ └── custom-policy_dev.json
└── requirements.txt
３．ドメイン駆動設計を適用してみる
それではDDD適用版のソースコードを解説していきたいと思います。ちなみに今回やったのは一般的に軽量DDDと呼ばれるもので、「ドメイン駆動設計のアーキテクチャを取り入れたプログラム構造を作る」ことを主目的としています。また伝統的なレイヤード（階層）アーキテクチャを採用しました。
３－１．パッケージ構成
今回のパッケージ構成は以下の通りです。共通部品（shared）を除くと４階層になっており、上から下へ処理が流れていく感じになります。
| No. | パッケージ（レイヤ） | 説明 |
|---|---|---|
| 1 | presentation | MVCモデルでいうところのコントローラを配置する層。外部（ブラウザ）からリクエストを受け取ってJSONで応答する役割。今回はここにChaliceのAPIエンドポイントを配置する。 |
| 2 | application | アプリケーションサービスを配置する層。ドメインとリポジトリを組み合わせて業務アプリケーションとしての機能を提供する。 |
| 3 | domain | ドメインモデル（エンティティや値オブジェクト）を配置する層。必要に応じ、ドメインサービスもここに配置。 |
| 4 | infrastracture | リポジトリを配置する層。DBやファイルシステムなど、インフラへアクセスするための機能を提供する。 |
| 5 | shared | その他、共通で使用するものはここに入れる。 |
３－２．概念図
少し端折っていますが、DDD適用後のクラス構成の概略はこのような感じです。
３－３．適用後のディレクトリ構成
ディレクトリ構成はこんな感じです。上記のクラスやパッケージなど、追加のモジュールは全てchalicelib配下に入れています。
product_search_sample/
├── app.py
├── .chalice
│ ├── config.json
│ └── custom-policy_dev.json
├── chalicelib
│ ├── application
│ │ ├── procuctAppricationSevice.py
│ │ ├── productSearchCommandForAppService.py
│ │ └── productSearchResult.py
│ ├── domain
│ │ └── models
│ │ └── product
│ │ └── products.py
│ ├── infrastracture
│ │ ├── baseRepositoryOpenSearch.py
│ │ ├── productRepositoryInterface.py
│ │ ├── productRepositoryOpenSearch.py
│ │ ├── productRepositoryStab.py
│ │ └── productSearchCommandForRepository.py
│ ├── presentation
│ │ └── productSearchController.py
│ └── shared
│ └── diModuleGenerator.py
└── requirements.txt
３－４．各パッケージの中身の解説
前段で説明したパッケージやクラスファイルなどの中身について解説していきます。DDDはボトムアップで進めるのが常套手段のようなので、説明もこれにならって domain → infrastracture → application → presentation の順に行っていきたいと思います。
３－４－１．domain パッケージ
ドメインモデルをここに配置しています。今回作成したのはエンティティのみで、値オブジェクトの利用はとりあえず見送っています。また今回は実装していませんが、ドメインサービスを使う場合はこの領域に放り込んでいく形になるかと思います。
（１）models.product.products.py
商品情報を保持するエンティティです。productクラスで商品１件単位の情報を取り扱い、productsクラスで複数の商品（product）をListで纏める感じにしています。
products.py
import dataclasses
from typing import List
@dataclasses.dataclass
class Product:
"""
DDDにおけるエンティティ
商品検索用のデータを保持する
本当は全てPrivate変数にしてgetterで取らせるようにしたいところ
"""
# 商品コード
productCode: str = None
# 商品名
productName: str = None
# ブランド
brand: str = None
# 性別
gender: str = None
# カテゴリ
category: str = None
# 価格
price: int = None
# 色
color: str = None
# サイズ
size: str = None
@dataclasses.dataclass
class Products:
"""
DDDにおけるエンティティ
商品検索結果一覧を保持する
"""
# product一覧（private変数）
__productList: List[Product] = dataclasses.field(default_factory=list)
def add(self, product: Product):
self.__productList.append(product)
def getProductList(self):
return self.__productList
def count(self):
return len(self.__productList)
３－４－２．infrastracture パッケージ
物理DB等へのアクセスを行うリポジトリクラスをここに配置しています。今回はOpenSearch検索用のリポジトリとテスト用リポジトリ（スタブ）の２つを用意しました。また、後の説明で出てくるDI（依存性の注入）を実現するために、Javaでいうところのインターフェース的なものも用意しています。
（１）productRepositoryInterface.py
リポジトリ用のインターフェースです。Pythonにはインターフェースという概念が存在しないようなので、抽象基底クラス（ABC）で代用しています。ここでは検索用のfindメソッドの実装を強制するようにしています。
productRepositoryInterface.py
import sys
import os
from abc import ABCMeta
from abc import abstractmethod
# 上位ディレクトリにパスを通す
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from domain.models.product.products import Product, Products
from infrastracture.productSearchCommandForRepository import ProductSearchCommandForRepository
class ProductRepositoryInterface(metaclass=ABCMeta):
"""
productRepositoryの基底クラス（インターフェースとして利用）
"""
@abstractmethod
def find(self, command: ProductSearchCommandForRepository) -> Products:
pass
（２）baseRepositoryOpenSearch.py
OpenSearch検索用のベースクラスです。コネクションの生成やクエリ発行用のメソッドを実装しており、実際にOpenSearchへの検索を行うクラスで継承することを想定しています。
baseRepositoryOpenSearch.py
import sys
import os
import traceback
import boto3
import json
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
class BaseRepositoryOpenSearch():
"""
OpenSearch（Elasticsearch）用リポジトリのベースクラス
"""
def __init__(self):
# Elasticsearch接続情報
# TODO: 設定ファイル（config.json）へ外出ししたいがまた今度
self.host = 'xxxxxx.ap-northeast-1.es.amazonaws.com'
self.port = 443
self.indexName = 'test-products'
# 認証情報取得
region = 'ap-northeast-1'
service = 'es'
credentials = boto3.Session().get_credentials()
self.awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
def _createConnection(self):
"""
protected関数
ESへ接続し、クライアントオブジェクトを返却する
"""
try:
return OpenSearch(
hosts = [{'host':self.host, 'port': self.port}],
http_auth = self.awsauth,
use_ssl = True,
verify_certs = True,
connection_class = RequestsHttpConnection
)
except Exception as e:
raise e
def _executeQuery(self, indexName, query):
"""
protected関数
クエリを実行し、結果を返却する
"""
try:
# ESへの接続とクエリ実行
esClient = self._createConnection()
return esClient.search(index=indexName, body=query)
except Exception as e:
raise e
finally:
esClient.close()
（３）productRepositoryOpenSearch.py
実際にOpenSearchへの検索を行うクラスです。インターフェース（ProductRepositoryInterface）とベースクラス（BaseRepositoryOpenSearch）の２つを継承しており、インターフェースで規定されたfindメソッドをここに実装することで検索が行えるようになります。戻り値としてドメインオブジェクトのエンティティ（Productsクラス）を返しています。
productRepositoryOpenSearch.py
import sys
import os
import traceback
import boto3
import json
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
# 上位ディレクトリにパスを通す
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from domain.models.product.products import Product, Products
from infrastracture.productSearchCommandForRepository import ProductSearchCommandForRepository
from infrastracture.productRepositoryInterface import ProductRepositoryInterface
from infrastracture.baseRepositoryOpenSearch import BaseRepositoryOpenSearch
class ProductRepositoryOpenSearch(ProductRepositoryInterface, BaseRepositoryOpenSearch):
"""
商品検索用のリポジトリ（Elasticsearch用）
"""
def find(self, command: ProductSearchCommandForRepository) -> Products:
"""
商品検索を行う
"""
if not command:
# できれば独自例外を返したい
raise ValueError("引数が空です")
try:
# 検索実行
query = self.__createSearchQuery(command)
searchResults = self._executeQuery(self.indexName, query)
# 戻り値（エンティティ）の組み立て
products = Products()
for result in searchResults["hits"]["hits"]:
source = result["_source"]
product = Product()
product.productCode = source["productCode"]
product.productName = source["productName"]
product.brand = source["brand"]
product.gender = source["gender"]
product.category = source["category"]
product.price = source["price"]
product.color = source["color"]
product.size = source["size"]
products.add(product)
return products
except Exception as e:
# スタックトレース出力
print(traceback.format_exc())
raise e
def __createSearchQuery(self, command: ProductSearchCommandForRepository) -> dict:
"""
private関数
OpenSearchに対して投げるクエリを生成する
"""
# ベースとなるクエリ
query = {
"from" : 0,
"size": 50,
"track_total_hits" : True,
"sort" : [
{"productName" : {"order" : "asc"}}
],
"query" : {
"bool" : {
"must" : []
}
}
}
# 検索条件の組み立て
# dataclassを一旦Dictに変換してキーと値を抜いている
for key in command.__dict__.keys():
searchParamValue = command.__dict__.get(key)
# 自分で定義したフィールドだけを検索条件の対象にする
if key.startswith("__"):
continue
if not searchParamValue:
continue
if key == "price":
# 検索条件がpriceの場合は数値での条件を指定
query["query"]["bool"]["must"].append(
{
"range" : {
key : {
"gte" : searchParamValue[0],
"lt" : searchParamValue[1]
}
}
}
)
else:
# price以外は文字列検索
query["query"]["bool"]["must"].append(
{
"terms" : {key : searchParamValue}
}
)
return query
（４）productRepositoryStub.py
テスト用のスタブクラスです。ダミー処理なのでインターフェース（ProductRepositoryInterface）だけを継承しています。とりあえず見ての通り、現時点では空のProductsクラスを返すのみです。ここはテストの要件に応じて処理を追加していくことになるかと思います。
productRepositoryStub.py
import sys
import os
# 上位ディレクトリにパスを通す
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from domain.models.product.products import Product, Products
from infrastracture.productSearchCommandForRepository import ProductSearchCommandForRepository
from infrastracture.productRepositoryInterface import ProductRepositoryInterface
class ProductRepositoryStab(ProductRepositoryInterface):
"""
リポジトリテスト用のスタブクラス
"""
def find(self, command: ProductSearchCommandForRepository) -> Products:
print("スタブクラスが実行されました！！")
return Products()
（５）productSearchCommandForRepository.py
OpenSearchに対して検索を行う際の、検索条件を保持するクラスです。リポジトリで実装されているfindメソッドの引数として渡します。入力チェックなんかが必要であればここに実装するといいかもしれません。
productSearchCommandForRepository.py
import dataclasses
from typing import List
@dataclasses.dataclass
class ProductSearchCommandForRepository:
"""
商品検索用の検索条件を保持するコマンドクラス
"""
# ブランド
brand: List[str] = None
# 性別
gender: List[str] = None
# 価格（小、大のList）
price: List[int] = None
# 色
color: List[str] = None
# サイズ
size: List[str] = None
def __post_init__(self):
"""
初期化処理（入力チェック）
"""
# TODO validation処理を書く
# if not self.brand:
# raise ValueError('ブランドが空です')
pass
３－４－３．applicationパッケージ
アプリケーションサービスをここに配置しています。前述のドメインとリポジトリをこの層で組み合わせることにより、業務アプリケーションとしての機能を実現することになります。
（１）procuctAppricationSevice.py
商品検索を行うためのサービスクラスです。searchProductsメソッド内でリポジトリに対して検索を行い、応答用オブジェクトを生成して返却しています。このクラスではDIコンテナによる依存性（リポジトリインスタンス）の注入が行えるように、コンストラクタの引数にリポジトリのインターフェース（ProductRepositoryInterface）を指定しています。DIコンテナにはPythonライブラリのinjectorを使ってみました。
procuctAppricationSevice.py
import sys
import os
from typing import List
from dataclasses import asdict
from injector import inject
# 上位ディレクトリにパスを通す
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from application.productSearchCommandForAppService import ProductSearchCommandForAppService
from application.productSearchResult import ProductSearchResult
from infrastracture.productRepositoryInterface import ProductRepositoryInterface
from infrastracture.productSearchCommandForRepository import ProductSearchCommandForRepository
class ProcuctAppricationSevice:
"""
商品検索用のアプリケーションサービス
"""
@inject
def __init__(self, productRepository: ProductRepositoryInterface) -> List[ProductSearchResult]:
# 外部ライブラリのinjectorを使って、Repositoryを外から注入する
self.productRepository = productRepository
def searchProducts(self, command: ProductSearchCommandForAppService):
# コマンドオブジェクトの組み換え
# とりあえずコピー元をDict化してアンパックして詰め替え。ここはファクトリにしたいところ
repositoryCommand = ProductSearchCommandForRepository(**asdict(command))
# 検索実行
products = self.productRepository.find(repositoryCommand)
# 応答用のオブジェクト組み立てと返却
productSearchResult = ProductSearchResult(products)
return productSearchResult
（２）productSearchCommandForAppService.py
検索条件を保持するクラスです。リポジトリでも同じようなものを定義していますが、層を跨いで使用するとメンテナンス性が悪くなる気もするのでアプリケーションサービス専用のクラスを用意してみました。
productSearchCommandForAppService.py
import dataclasses
from typing import List
@dataclasses.dataclass
class ProductSearchCommandForAppService:
"""
商品検索用の検索条件を保持するコマンドクラス
"""
# ブランド
brand: List[str] = None
# 性別
gender: List[str] = None
# 価格（小、大のList）
price: List[int] = None
# 色
color: List[str] = None
# サイズ
size: List[str] = None
def __post_init__(self):
"""
初期化処理（入力チェック）
"""
# TODO validation処理を書く
# if not self.brand:
# raise ValueError('ブランドが空です')
pass
（３）productSearchResult.py
アプリケーションサービスからの応答データを保持するクラスです。ドメインオブジェクトであるProductsクラスをラッピングし、createApiResponseメソッド経由でAPI応答用のJSONを取得できるようにしています。ただこのJSON生成処理については、正直ここに入れるべきかどうか悩みどころです。責務的にはpresentation層で実装してもいい気もします。
あとここに実装するとしても、createApiResponseメソッドを親クラスに外出しして継承するような作りにすべきですが、疲れたのでこの辺はサボってしまいました。
productSearchResult.py
import sys
import os
import json
import dataclasses
from typing import List
# 上位ディレクトリにパスを通す
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from domain.models.product.products import Product, Products
@dataclasses.dataclass
class ProductSearchResult:
"""
Applicationレイヤ用のデータクラス
DDDのエンティティをラッピングしてAPI（プレゼンテーション層）向けの応答データとする
"""
# ドメインオブジェクト private変数
__products: Products = None
def createApiResponse(self):
"""
API応答用JSONを生成する
"""
# 最終的な応答値の雛形を定義
responsData = {
"status" : "success",
"totalCount" : 0,
"results" : [],
}
# 検索結果が存在しない場合は早々に処理終了
totalCount = self.__products.count()
if totalCount == 0:
return responsData
# 応答値の生成
responsData["totalCount"] = totalCount
for product in self.__products.getProductList():
responsData["results"].append(
{
"productCode" : product.productCode,
"productName" : product.productName,
"brand" : product.brand,
"gender" : product.gender,
"category" : product.category,
"price" : product.price,
"color" : product.color,
"size" : product.size,
}
)
return json.dumps(responsData, ensure_ascii=False)
３－４－４．presentation パッケージ
MVCモデルでいうところのコントローラを配置する層です。今回はフレームワークとしてChaliceを使用してるので、REST APIのエンドポイント定義を記載したChaliceモジュールをここに格納しています。
（１）productSearchController.py
前述の通り、ChaliceのAPIエンドポイントを定義したモジュールです。リクエストから検索条件を取得して、アプリケーションサービス経由で検索を実行し、結果をJSON形式で返却します。また、アプリケーションサービスのインスタンスを生成する際にDIコンテナ（前述のinjectorライブラリ）を使用しています。
なお今回はChaliceのBlueprints機能を使ってapp.pyを分割してみました。Blueprintsの詳細についてはこちらの記事を参照ください。
productSearchController.py
import sys
import os
import json
import traceback
from chalice import Response
from chalice import Blueprint
from injector import Injector, Module
# 上位ディレクトリにパスを通す
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from application.procuctAppricationSevice import ProcuctAppricationSevice
from application.productSearchCommandForAppService import ProductSearchCommandForAppService
from shared.diModuleGenerator import DiModuleGenerator
# ブループリント初期化
extraChaliceApp = Blueprint(__name__)
# CloudFrontの制約で、GETではリクエストボディが使えないためPOSTにする
@extraChaliceApp.route('/searchProducts', methods=['POST'], cors=True, api_key_required=False)
def searchProducts():
"""
商品検索処理のエンドポイント
"""
try:
# リクエストボディからcommandを生成
requestBody = extraChaliceApp.current_request.json_body
command = ProductSearchCommandForAppService(**requestBody)
# DIコンテナからアプリケーションサービスを取得してコール
diModule = DiModuleGenerator.generatDiModule(DiModuleGenerator.OPENSEARCH)
injector = Injector([diModule])
service = injector.get(ProcuctAppricationSevice)
productSearchResult = service.searchProducts(command)
# API応答値の返却
return Response(
body = productSearchResult.createApiResponse(),
headers = {'Content-Type': 'application/json'},
status_code = 200
)
except Exception as e:
# スタックトレース出力とエラー応答
print(traceback.format_exc())
responseData = {"message" : "内部エラーが発生しました"}
return Response(
body = json.dumps(responseData, ensure_ascii=False),
headers = {'Content-Type': 'application/json'},
status_code = 500
)
３－４－５．shared パッケージ
こちらが最後のパッケージです。今までの説明に含まれないような、全体で共用するクラスなどを配置します。
（１）diModuleGenerator.py
DIコンテナ用の定義クラスです。同じファイル内に３つほどクラスがあり、OpensearchDiModuleクラスにはOpenSearch用のリポジトリ定義を、StubDiModuleクラスにはスタブ用のリポジトリ定義を記載しています。実際のクラスの生成はDiModuleGenerator.generatDiModule()のクラスメソッドで行うようにしており、引数によってOpenSearchかスタブかを切り替えられる形で実装しています。
diModuleGenerator.py
import sys
import os
from injector import Injector, Module
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from infrastracture.productRepositoryInterface import ProductRepositoryInterface
from infrastracture.productRepositoryOpenSearch import ProductRepositoryOpenSearch
from infrastracture.productRepositoryStab import ProductRepositoryStab
class OpensearchDiModule(Module):
"""
DIする内容を定義するクラス（OpenSearch）
"""
def configure(self, binder):
binder.bind(ProductRepositoryInterface, to=ProductRepositoryOpenSearch)
class StubDiModule(Module):
"""
DIする内容を定義するクラス（テスト用のスタブ）
"""
def configure(self, binder):
binder.bind(ProductRepositoryInterface, to=ProductRepositoryStab)
class DiModuleGenerator():
"""
DI定義を生成するクラス
"""
OPENSEARCH = 1
STAB = 2
@classmethod
def generatDiModule(cls, arg):
if arg == cls.OPENSEARCH:
return OpensearchDiModule()
elif arg == cls.STAB:
return StubDiModule()
else:
# デフォルトはスタブを返す
return StubDiModule()
４．さいごに
長くなりましたが、解説は以上となります。この程度の処理にDDDを適用すると逆にコードが増えて扱いづらい気もしますし、そもそも検索APIにDDDを使うのはどうなの？という話もありますが、まあ今回の例は練習ということで。。
なお今回はセオリーに従って実装しているつもりですが、DDDとして正しいかどうかについては模索中の状態です。また私は元々がJava屋で、Pythonに関しては現在進行形で走りながら覚えている感じなので、ソースコードに微妙なところがあったり用語がJava寄りになってしまったりと若干不備があるかもしれません。もし微妙な点や改善点などあれば指摘頂けると助かります。
この記事が誰かのお役に立てると幸いです。
Discussion