■ はじめに

少し前にドメイン駆動設計の本を読んでかなり勉強になったので、自分なりの理解をまとめておきます。

この記事に書いてあるコードはこちらのリポジトリにまとまっています。

■ ドメイン駆動設計とは

ドメインとは

ソフトウェア開発におけるドメイン(領域)とは「プログラムを適用する対象となる領域」を指します。
例えば、「物流システム」であれば、倉庫・貨物・輸送手段などがドメイン(領域)に含まれますし、「会計システム」であれば金銭・帳票などがドメインに含まれるでしょう。

つまり、ドメインに何が含まれるかはシステムが何を対象とするか(物流、会計など)によって大きく変化します。

これから作るシステムが何を対象として、どんな問題を解決するものかによって実装するクラスも変わってくると思います。当然ですね。

ドメイン駆動設計とは

ドメイン駆動設計とは、システムが対象とするドメイン(領域)を正しく理解し、問題解決に必要な要素を抽出・抽象化しソースコードに落とし込むことです。

回りくどい言い方をしていますが、当たり前ですね。物流システムを作るには 貨物クラス や 倉庫クラス が必要なことは誰でも想像がつきそうです。

ドメインモデルとは

ドメイン駆動設計では「システムが解決すべき問題に必要な要素を抽出・抽象化」します。
対象ドメインから必要な要素の必要な特徴のみを抽出・抽象化した概念がドメインモデルです。

モデルとは現実の事象・概念を抽象化した概念です。

例えば、物流システムにおける「倉庫」を考えてみます。
現実の倉庫は、建物の広さ、高さ、壁の材質、扉の数、、、etc と無数の要素がありますが、こと物流システムにおいて、これらの特徴は必要ないはずです。倉庫に保管されている荷物や商品がわかれば十分なはずです。

この、「保管されている荷物や商品のみ管理できる倉庫」という概念が物流ドメインにおける倉庫のドメインモデルです。

ドメインオブジェクトとは

ドメインモデルはあくまでも抽象化された概念です。このドメインモデルをモジュールとしてソースコードに落とし込んだものがドメインオブジェクトです。

つまり

|ドメイン| -抽象化-> |ドメインモデル| -ソースコード化-> |ドメインオブジェクト|

ドメイン駆動設計のパターン

ドメイン駆動設計にはいくつかの要素があり、「知識を表現するもの」と「アプリケーションを実現するもの」に分類されるます。

<> 知識を表現するパターン

値オブジェクト
ドメインの知識を表現するオブジェクト
エンティティ
ドメインの概念を表現するオブジェクト
ドメインサービス
値オブジェクトやエンティティではうまく表現できない知識を取り扱うためのパターン

<> アプリケーションを実現するためのパターン

リポジトリ
データの保存や復元といった永続化や再構築を担当するオブジェクト
アプリケーションサービス
値オブジェクト・エンティティ・ドメインサービス・リポジトリの4要素を協調動作させ、アプリケーションとして成り立たせる
ファクトリ
オブジェクトを作る知識に特化したオブジェクト

<> 知識を表現するより発展的なパターン

集約
整合性を保つ協会。値オブジェクトやエンティティといったドメインオブジェクトを束ねて複雑なドメイン概念を表現する。
仕様
オブジェクトがある特定の条件下にあるかを評価するモジュール

■ 値オブジェクト

値オブジェクトはシステム固有の値を表現するために定義されたオブジェクトです。
値オブジェクトは属性によって識別されるオブジェクトで、属性が同じであるならば同じオブジェクトとみなされます。 (逆にエンティティはIDなどで同一性を識別する)

値オブジェクトにすべきかどうかは「そこにルールが存在しているか」「それ単体で扱いたいか」を基準にすると良い。

特徴

不変(イミュータブル)なオブジェクト
オブジェクト固有のふるまいを実装し、ロジックの散在を防止 (等価性の実装)
バリデーションで不正な値を存在させない

値オブジェクトを使うモチベーション

表現力を増やす
値がどのような要素で成り立っているかがわかりやすい
不正な値を存在させない
バリデーションを実装できる
誤った代入を防ぐ
イミュータブルなオブジェクトとして定義できる
ロジックの散在を防ぐ
メソッドとしてオブジェクトの振る舞いを定義することでそのオブジェクトに関するロジックをまとめる

実装例

ユーザー名を表現する値オブジェクトを実装してみます。

<> dataclassを使った実装

不変なオブジェクト
eqの実装(自動)

value_object_1.py

# dataclasses: https://docs.python.org/ja/3/library/dataclasses.html
from dataclasses import dataclass

# dataclasses: https://docs.python.org/ja/3/library/dataclasses.html
# frozen=True: オブジェクトを不変にする
# eq=True: __eq__ を自動実装する (デフォルトTrue)
@dataclass(frozen=True)
class UserName:
  first_name: str
  last_name: str
  
  def __str__(self):
    return f"{self.first_name} {self.last_name}"

if __name__ == "__main__":
  user_name1 = UserName(first_name="keita", last_name="midorikawa")
  user_name2 = UserName(first_name="keita", last_name="midorikawa")
  # user_name1.first_name = "hoge"  # エラー
  print(user_name1)
  print(user_name1 == user_name2)

<> pydanticを使った実装

不変なオブジェクト
eqの実装(自動)
バリデーション

value_object.py

from pydantic import BaseModel, ValidationError, validator

# pydantic: https://pydantic-docs.helpmanual.io/usage/models/
class UserName(BaseModel):
    first_name: str
    last_name: str

    # pydantic - Validators: https://pydantic-docs.helpmanual.io/usage/validators/
    @validator("first_name", "last_name")
    def name_validator(cls, v):
        if len(v) <= 0 or len(v) > 32:
            raise ValueError("error..")
        return v

    def __str__(self):
        return f"{self.first_name} {self.last_name}"

    class Config:
        # イミュータブルなオブジェクトにする
        # pydantic - ModelConfig - Options: https://pydantic-docs.helpmanual.io/usage/model_config/#options
        allow_mutation = False

if __name__ == "__main__":
    user_name1 = UserName(first_name="keita", last_name="midorikawa")
    user_name2 = UserName(first_name="keita", last_name="midorikawa")
    # user_name3 = UserName(first_name="", last_name="")    # バリデーションエラー
    # user_name1.first_name = "hoge"    # エラー
    print(user_name1)    # keita midorikawa
    print(user_name1 == user_name2)    # True

■ エンティティ

エンティティはドメインモデルを実装したドメインオブジェクトです。
エンティティはID(同一性)によって識別され、値オブジェクトのように属性が同じだからといって同一のオブジェクトとはなりません。

例えば、「ユーザー」はエンティティです。「ユーザー」はIDによって識別され、同姓同名だからといって同一のオブジェクトとして扱われるわけではないからです。

「ユーザー」のようにエンティティに分類されるオブジェクトは、作成〜削除までのライフサイクルが存在します。逆にライフサイクルが存在しないオブジェクトは値オブジェクトに分類します。

特徴

可変(ミュータブル)である
ユーザーの年齢や身長といった属性が変化するのと同じように
属性が同じであっても区別される
ID(同一性)により区別される

実装例

ユーザーを表現するentityを実装してみます。

entity.py

from value_object import UserName
from pydantic import BaseModel

class User(BaseModel):
    id: int
    user_name: UserName

    def change_name(self, user_name: UserName):
        """外部から直接インスタンス変数を変更させてはいけない (デメテルの法則)"""
        self.user_name = user_name

    def __eq__(self, other):
        if other is None or not isinstance(other, User):
            return False
        return self.id == other.id

if __name__ == "__main__":
    user1 = User(id=1, user_name=UserName(first_name="keita", last_name="midorikawa"))
    user2 = User(id=2, user_name=UserName(first_name="keita", last_name="midorikawa"))
    user3 = User(id=1, user_name=UserName(first_name="makoto", last_name="midorikawa"))

    # IDによってい識別される
    print(user1 == user2)  # False
    print(user1 == user3)  # True

    # ミュータブル
    user1.change_name(UserName(first_name="hoge", last_name="fuga"))
    print(user1)  # True

■ ドメインサービス

値オブジェクトやエンティティとして記述すると不自然になってしまう「操作」を実装するのがドメインサービスです。
複数のドメインオブジェクト(値オブジェクト・エンティティ)を横断するような操作で使うことが多いです。

しかしながら、ドメインサービスが肥大すると、ドメインオブジェクトと処理の関連が見えづらくなるので、「ふるまい」はなるべく値オブジェクト・エンティティに実装するのが望ましいです。


class User:
    # ... 略 ...
    def exists(self, user: User):
      # 重複を確認するコード
      
if __name__ == "__main__":
  user = User(id=1, ...)
  # 自分自身の存在確認を自分自身にしているかのようなコードになってしまう。
  user.exists(user)

特徴

複数のドメインオブジェクトを横断する処理など、値オブジェクトやエンティティとして記述すると不自然になってしまう操作を実装する
副作用のあるインスタンス変数を持たない
可能な限り入出力(DB, ファイル)やインフラとのやり取りが伴う処理は扱わない(後述のリポジトリで扱う)

実装例

<> 例1

エンティティにユーザーの存在確認をさせるのは不自然なので、ドメインサービスにユーザーの存在確認を行うメソッドを実装します。

※ 個人的に存在確認の処理はリポジトリに実装してもいいと思うが。。。

domain_service_1.py

from value_object import UserName
from entity import User
from repository import IUserRepository, UserRepository

class UserService:
    """ドメインサービス
    データの入出力はリポジトリを利用することで実現する。
    後でテストができるように、リポジトリのインターフェースに依存させる
    """
    def __init__(self, user_repository: IUserRepository):
        # インターフェース依存
        self.user_repository = user_repository
    
    # 「ユーザーの重複確認」はドメインのルールに近いので、existsはドメインサービスに実装する
    def exists(self, user: User) -> bool:
        # データの入出力処理をリポジトリに閉じ込めることで、見通しが良くなる
        found = self.user_repository.find_by_name(user.user_name)
        return found is not None

if __name__ == "__main__":
    # リポジトリ
    store_path = "store.json"
    repository = UserRepository(store_path)

    # ドメインサービス
    service = UserService(repository)

    # 初期化
    repository.clear()

    # 同名のユーザーが存在しなければuserを追加
    user = User(id=0, user_name=UserName(first_name="kta", last_name="mido"))
    if not service.exists(user):
        user = repository.save(user)

    # userの存在確認
    print(service.exists(user))  # True

    # userの削除
    repository.delete(user)

    # userの存在確認
    print(service.exists(user))  # False

<> 例2

入出力やインフラとのやり取りを伴わないドメインサービスの例として、物流システムにおける、物流拠点から物流拠点への輸送を考えてみます。

| 物流拠点 |  -- 輸送 --> | 物流拠点 |

domain_service_2.py

class PhysicalDistributionBase:
    """物流拠点を表すエンティティ"""
    def __init__(self):
        self.baggages = set()

    def ship(self, baggage):
        self.baggages.remove(baggage)

    def receive(self, baggage):
        self.baggages.add(baggage)


class TransportService:
    """輸送の振る舞いを定義するドメインサービス"""
    def transport(self, src: PhysicalDistributionBase, dst: PhysicalDistributionBase, baggage):
        src.ship(baggage)
        dst.receive(baggage)

if __name__ == "__main__":

    # 物流拠点を用意
    base1 = PhysicalDistributionBase()
    base2 = PhysicalDistributionBase()

    # 荷物を base1 に登録
    baggage = "b1"
    base1.receive(baggage)

    # base1 -> base2 に baggage を輸送
    transport_service = TransportService()
    transport_service.transport(base1, base2, baggage)

■ リポジトリ

リポジトリとはデータの永続化や再構築を行うオブジェクトで、インスタンスの保存や復元を担当します。

データの書き込み・読み込みといった操作は処理の本質をぼやけさせてしまうので、ビジネスロジックから分離させる必要があります。ビジネスロジックはデータストアが何であるかを感知すべきではありません。

ビジネスロジックが特定のインフラに依存するとソフトウェアが硬直化してしまいます。

特徴

データの永続化や再構築を行うオブジェクト
ビジネスロジック側がデータストアのバックエンドを意識しなくていいように実装する

永続化

基本的に永続化のふるまいは永続化を行うオブジェクトを引数に取ります。

def save(self, user: User):
    # ...略...

def delete(self, user: User):
    # ...略...

エンティティの識別子と更新項目を引き渡して「更新」させるようなメソッドはリポジトリには含めるべきではありません。エンティティが保持するデータの変更はエンティティ自身に実装するべきです。

同様にエンティティを生成する処理もリポジトリには含めません。エンティティはコンストラクタから直接生成します。コンストラクタ以外から生成する場合はファクトリ(後述)を利用します。

# これらはリポジトリに実装しない

def update_name(id, name):
    # ...略...

def update_address(id, address):
    # ...略...

実装例

ユーザーの登録・取得ができるリポジトリを実装します。

リポジトリを利用する場合は、利用する側がリポジトリのインターフェースに依存するように実装します。 (モック実装によるテストが容易にするため)

repository.py

import abc
import os
from typing import Optional, List
from entity import User
from value_object import UserName
from pydantic import BaseModel

class UserStoreSchema(BaseModel):
    """データストアの構造定義"""
    id: int = 1
    users: List[User]

class IUserRepository(metaclass=abc.ABCMeta):
    """リポジトリのインターフェース"""
    @abc.abstractmethod
    def save(self, user: User) -> User:
        raise NotImplementedError

    @abc.abstractmethod
    def delete(self, user: User):
        raise NotImplementedError

    @abc.abstractmethod
    def find_by_name(self, user_name: UserName) -> Optional[User]:
        raise NotImplementedError

    @abc.abstractmethod
    def find(self, id: int) -> Optional[User]:
        raise NotImplementedError

class UserRepository(IUserRepository):
    """リポジトリ
    データの入出力処理を担当
    """
    def __init__(self, store_path: str):
        self._store_path = store_path

    def _save(self, schema: UserStoreSchema):
        with open(self._store_path, "w") as writer:
            writer.write(schema.json(ensure_ascii=False, indent=2))

    def _load(self) -> UserStoreSchema:
        if os.path.exists(self._store_path):
            return UserStoreSchema.parse_file(self._store_path)
        return UserStoreSchema(users=[])

    def clear(self):
        self._save(UserStoreSchema(users=[]))

    def save(self, user: User) -> User:
        store = self._load()
        users = list(filter(lambda e: e.id == user.id, store.users))
        if len(users) > 0:
            # update
            delete_user = users[0]
            store.users.remove(delete_user)
        store.users.append(user)
        self._save(store)
        return user

    def delete(self, user: User):
        store = self._load()
        users = list(filter(lambda e: e.id == user.id, store.users))
        if len(users) > 0:
            delete_user = users[0]
            store.users.remove(delete_user)
            self._save(store)

    def find_by_name(self, user_name: UserName) -> Optional[User]:
        for user in self._load().users:
            if user.user_name == user_name:
                return user
        return None

    def find(self, id: int) -> Optional[User]:
        for user in self._load().users:
            if user.id == id:
                return user
        return None


if __name__ == "__main__":
    store_path = "store.json"

    # リポジトリ
    repository = UserRepository(store_path)

    # 初期化
    repository.clear()

    # ユーザー追加処理
    username = UserName(first_name="kta", last_name="mido")
    user = User(id=0, user_name=username)
    user = repository.save(user)

    print(repository.find_by_name(username))

    # userの削除
    repository.delete(user)

<> テスト

テスト用のインメモリなリポジトリを実装してドメインサービスのテストを行ってみます。

UserService は IUserRepository(インターフェース) に依存しているため、 IUserRepository を実装したクラスであれば何でも受け取ることができます。

repository_test.py

from typing import Optional
from entity import User
from value_object import UserName
from repository import IUserRepository, UserStoreSchema
from domein_service import UserService

class InMemoryRepository(IUserRepository):
    """テスト用のインメモリなリポジトリを実装"""

    def __init__(self):
        self.data = UserStoreSchema(users=[])

    def save(self, user: User) -> Optional[User]:
        users = list(filter(lambda e: e.id == user.id, self.data.users))
        if len(users) > 0:
            # update
            delete_user = users[0]
            self.data.users.remove(delete_user)
        self.data.users.append(user)

    def find_by_name(self, user_name: UserName) -> Optional[User]:
        for user in self.data.users:
            if user.user_name == user_name:
                return user
        return None

    def delete(self, user: User):
        pass

    def find(self, id: int) -> Optional[User]:
        return None



class UserServiceTest:
    """ドメインサービスのテストコード"""

    def test_exists_false(self):
        """UserServiceのexistsメソッドをテスト"""
        user = User(id=1, user_name=UserName(first_name="kta", last_name="mido"))
        repository = InMemoryRepository()
        service = UserService(repository)
        assert service.exists(user) == False

    def test_exists_true(self):
        """UserServiceのexistsメソッドをテスト"""
        user = User(id=1, user_name=UserName(first_name="kta", last_name="mido"))
        repository = InMemoryRepository()
        repository.save(user)
        service = UserService(repository)
        assert service.exists(user) == True


if __name__ == "__main__":
    test = UserServiceTest()

    test.test_exists_false()
    test.test_exists_true()

■ アプリケーションサービス

アプリケーションサービスはドメインオブジェクトを協調させてユースケースを実現します。

ドメインオブジェクトを組み合わせ他一連の処理。つまりスクリプトのようなものです。

アプリケーションサービスはあくまでもドメインオブジェクトの「利用」に徹するべきで、ドメインのルールを記述すべきではありません。

例えばユーザーの重複判定といった「ドメインのルール」はドメインサービス( UserService.exists )に実装して、重複の定義が変わったときに UserService.exists のみを変更すればいいようにすべきです。

特徴

アプリケーションサービスはドメインオブジェクトを組み合わせて実行するスクリプトのようなもの
※ あくまで、ドメインオブジェクトの「利用」のみで、ルールを記述してはいけない
アプリケーションサービスはメソッドのふるまいを変化させる状態(インスタンス変数)を持たない

実装例

ユーザーCreate, Read, Update, Deleteを行うアプリケーションサービスを実装してみます。

application.py

import abc
from typing import Optional
from value_object import UserName
from entity import User
from repository import UserRepository, IUserRepository
from domein_service import UserService

class IUserApplicationService(metaclass=abc.ABCMeta):
    """アプリケーションサービスのインターフェース
    クライアント側でモック実装を作れるようにインターフェースを作る
    """
    @abc.abstractmethod
    def register(self, id: int, first_name: str, last_name: str):
        raise NotImplementedError

    @abc.abstractmethod
    def get(self, id: int) -> Optional[User]:
        raise NotImplementedError

    @abc.abstractmethod
    def update(self, id: int, first_name: Optional[str] = None, last_name: Optional[str] = None): 
        raise NotImplementedError

    @abc.abstractmethod
    def delete(self, id: int):
        raise NotImplementedError


class UserApplicationService(IUserApplicationService):
    def __init__(self, repository: IUserRepository, service: UserService):
        self.service = service
        self.repository = repository

    def register(self, id: int, first_name: str, last_name: str):
        user_name = UserName(first_name=first_name, last_name=last_name)
        user = User(id=id, user_name=user_name)
        if self.service.exists(user):
            raise Exception(f"{user.user_name} はすでに存在しています")
        self.repository.save(user)

    def get(self, id: int) -> Optional[User]:
        """
        本が言うには、ドメインオブジェクト(User)のふるまいの呼び出しはアプリケーションサービスの役目であり、クライアントが自由に操作できてはいけないらしい。
        なので取得したオブジェクトが変更できないようにDTO(Data Transfer Object)という参照専用の型に変換して返すのが本当はいいらしい
        ```
        # データ転送用オブジェクト(DTO)
        class UserData:
            def __init__(self, user: User):
                self.id = user.id
                self.user_name = str(user.user_name)
        ```
        """
        return self.repository.find(id)

    def update(self, id: int, first_name: Optional[str] = None, last_name: Optional[str] = None): 
        """
        UserUpdateCommandのようなコマンドオブジェクトを作成して、引数をオブジェクトとして受け取ってもよい。
        ```
        @dataclass(frozen=True)
        class UserUpdateCommand(BaseModel):
            id: int
            first_name: Optional[str]
            last_name: Optional[str]
        ```
        """
        user = self.repository.find(id)
        if user is None:
            raise Exception(f"ユーザーが見つかりません (id={id})")
        if first_name is not None and last_name is not None:
            user.user_name = UserName(first_name=first_name, last_name=last_name)
            if (self.service.exists(user)):
                raise Exception(f"{user.user_name} はすでに存在しています")
        self.repository.save(user)
    
    def delete(self, id: int):
        user = self.repository.find(id)
        if user is None:
            raise Exception(f"ユーザーが見つかりません (id={id})")
        self.repository.delete(user)



if __name__ == "__main__":
    store_path = "store.json"

    # リポジトリ
    repository = UserRepository(store_path)

    # ドメインサービス
    service = UserService(repository)

    # 初期化
    repository.clear()

    # アプリケーション
    app = UserApplicationService(repository, service)

    # 追加
    app.register(1, "kta", "mido")
    app.register(2, "foo", "bar")
    app.register(3, "hoge", "piyo")

    # 取得
    print(app.get(1))  # id=1 user_name=UserName(first_name='kta', last_name='mido')

    # 更新
    app.update(1, "keita", "midorikawa")

    # 削除
    app.delete(3)

凝集度

凝集度はモジュールの責任範囲がどれだけ集中しているかを測る尺度で、高いほど堅牢性・信頼性・再利用性・可読性の観点から好まし意図されています。

「凝集度が高い」というのは「インスタンス変数がすべてのメソッドで利用されている」ということになります。

凝集度を測る方法にLCOM(Lack of Cohesion in Methods) という計算式があります。

<> 凝集度が低い例

v1,v2はmethod_aでのみ、v3,v4はmethod_bでのみ利用されています。

class A:
  def __init__(self, v1, v2, v3, v4):
      self.v1 = v1
      self.v2 = v2
      self.v3 = v3
      self.v4 = v4
  
  def method_a(self):
      return self.v1 + self.v2
  
  def method_b(self):
      return self.v3 + self.v4

<> 凝集度が高い例

すべてのインスタンス変数が全てのクラスで利用されています。

class A:
  def __init__(self, v1, v2):
      self.v1 = v1
      self.v2 = v2
  
  def method_a(self):
      return self.v1 + self.v2

class B:
  def __init__(self, v3, v4):
      self.v3 = v3
      self.v4 = v4
  
  def method_b(self):
      return self.v3 + self.v4

<> 凝集度を高めたアプリケーションサービスの実装

アプリケーションサービスの実装例のコードの凝集度を高めてみます。
※ 個人的にはここまでやる必要あるか、、、って思います

application_high_cohesion.py

import abc
from typing import Optional
from value_object import UserName
from entity import User
from repository import UserRepository, IUserRepository
from domein_service import UserService

class IUserRegisterApplicationService(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def register(self, id: int, first_name: str, last_name: str):
        raise NotImplementedError

class IUserGetApplicationService(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def get(self, id: int) -> Optional[User]:
        raise NotImplementedError

class IUserUpdateApplicationService(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def update(self, id: int, first_name: Optional[str] = None, last_name: Optional[str] = None): 
        raise NotImplementedError

class IUserDeleteApplicationService(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def delete(self, id: int):
        raise NotImplementedError

class UserRegisterApplicationService(IUserRegisterApplicationService):
    def __init__(self, repository: IUserRepository, service: UserService):
        self.service = service
        self.repository = repository

    def register(self, id: int, first_name: str, last_name: str):
        user_name = UserName(first_name=first_name, last_name=last_name)
        user = User(id=id, user_name=user_name)
        if self.service.exists(user):
            raise Exception(f"{user.user_name} はすでに存在しています")
        self.repository.save(user)

class UserGetApplicationService(IUserGetApplicationService):
    def __init__(self, repository: IUserRepository, service: UserService):
        self.service = service
        self.repository = repository

    def get(self, id: int) -> Optional[User]:
        return self.repository.find(id)

class UserUpdateApplicationService(IUserUpdateApplicationService):
    def __init__(self, repository: IUserRepository, service: UserService):
        self.service = service
        self.repository = repository

    def update(self, id: int, first_name: Optional[str] = None, last_name: Optional[str] = None): 
        user = self.repository.find(id)
        if user is None:
            raise Exception(f"ユーザーが見つかりません (id={id})")
        if first_name is not None and last_name is not None:
            user.user_name = UserName(first_name=first_name, last_name=last_name)
            if (self.service.exists(user)):
                raise Exception(f"{user.user_name} はすでに存在しています")
        self.repository.save(user)

class UserDeleteApplicationService(IUserDeleteApplicationService):
    def __init__(self, repository: IUserRepository):
        self.repository = repository

    def delete(self, id: int):
        user = self.repository.find(id)
        if user is None:
            raise Exception(f"ユーザーが見つかりません (id={id})")
        self.repository.delete(user)



if __name__ == "__main__":
    store_path = "store.json"

    # リポジトリ
    repository = UserRepository(store_path)

    # ドメインサービス
    service = UserService(repository)

    # 初期化
    repository.clear()

    # アプリケーション
    register_app = UserRegisterApplicationService(repository, service)

    # 追加
    register_app.register(1, "kta", "mido")
    register_app.register(2, "hoge", "fuga")
    register_app.register(3, "foo", "bar")

    # 取得
    get_app = UserGetApplicationService(repository, service)
    print(get_app.get(1))  # id=1 user_name=UserName(first_name='kta', last_name='mido')

    # 更新
    update_app = UserUpdateApplicationService(repository, service)
    update_app.update(1, "keita", "midorikawa")

    # 削除
    delete_app = UserDeleteApplicationService(repository)
    delete_app.delete(1)

■ 柔軟性をもたらす依存関係のコントロール

依存とは

<> 参照による依存関係

ObjectAがObjectBを参照している関係です。
依存する側のモジュールから依存される側のモジュールに矢印を伸ばして表現します。


class ObjectA:
    obj_b: ObjectB

<> 汎化による依存関係

いわゆるインターフェースの実装や継承といった関係です。
実装クラスからインターフェースに矢印を伸ばして表現します。

白抜きの矢印は汎化を示します。

import abc

class IUserRepository(metaclass=abc.ABCMeta):
    @abc.abstractclassmethod
    def save(self, user: User) -> User:
        raise NotImplementedError

class UserRepository(IUserRepository):
    def save(self, user: User) -> User:
        pass

抽象クラスに依存せよ

<> 依存関係逆転の原則

上位レベルのモジュールは下位レベルのモジュールに依存してはならない、どちらのモジュールも抽象に依存するべき
抽象は実装の詳細に依存してはならない、実装の詳細が抽象に依存するべき

<> プログラムのレベル

プログラムには レベル という概念があります。

低レベル (下位レベル)
入出力から近く、機械に近い具体的な処理 (例えばRepository)
高レベル (上位レベル)
入出力から遠く、人間に近い抽象的な処理 (例えばApplicationService)

<> 依存関係の変化

IUserRepository (インターフェース)を導入することで、 ApplicationService (上位)から Repository (下位)への依存がなくなり、どちらも抽象への依存になります。
こうすることで、ApplicationService のテストでテスト用のモッククラスなどを利用できるようになります。

ServiceLocatorパターン

ServiceLocatorオブジェクトに依存解決先となるオブジェクトを事前に登録しておいて、インスタンスが必要となる各所でServiceLocatorを経由してインスタンスを取得する依存解決のパターンです。
ServiceLocator経由でインスタンスを生成することで、InMemoryUserRepositoryやUserRepositoryなどの具象クラスのインスタンス化を行うコードが点在しなくなり、リポジトリの具象クラスの差し替えがしやすくなります。

※ ただし、ServiceLocatorパターンは下記の理由でアンチパターンとも言われいます

依存関係が外部から見えづらくなる
resolveよりも先にregisterが呼び出されている必要があるため、コード自体が副作用を持つようになる。(場所によってインスタンス化が成功したり失敗したりする)
テストの維持が難しくなる

※ 個人的には、副作用とソースを追いづらくなるデメリットがあるので使わないほうが良いと思います。グローバル変数に状態を持たせておくのと何ら変わらないので。

<> 実装例

Service Locator パターンについて

IUserRepositoryを解決するServiceLocatorを実装してみます。

service_locator.py

from typing import Type, Dict, Any
from repository import UserRepository, IUserRepository
from entity import User
from value_object import UserName

class ServiceLocator:
    services: Dict[Any, Any] = {}

    @classmethod
    def register(cls, interface: Type, type: Type):
        cls.services[interface] = type

    @classmethod
    def resolve(cls, interface: Type) -> Type:
        return cls.services[interface]

if __name__ == "__main__":
    # インターフェースに紐づく形で利用するリポジトリの具象クラスを登録しておく
    ServiceLocator.register(IUserRepository, UserRepository)

    # リポジトリを利用するときにServiceLocator経由でリポジトリの具象クラスを取得
    Repository = ServiceLocator.resolve(IUserRepository)

    # リポジトリインスタンス
    repository = Repository("store.json")

    user = User(id=1, user_name=UserName(first_name="foo", last_name="bar"))
    repository.save(user)

    print(repository.find(1))

    repository.delete(user)

IoC Container(DI Container)パターン

<> Dependency Injector

Pythonでやるならこれですかね。

インストール

pip install dependency-injector

プロバイダー

よく使いそうなのはこの辺？

providers.Factory
呼び出しごとに新しいインスタンスを作る
providers.Singleton
いつ呼び出されても同じインスタンスを返す
provider.Callable
関数呼び出しを行う
provider.Config
コンフィグを保持するプロバイダ
provider.Resource
よーわからんけど初期化とshutdownを行う？

<> 実装

UserApplicationServiceのDIコンテナを作ってみます。

di_container.py

from dependency_injector import containers, providers
from repository import UserRepository
from domein_service import UserService
from application import UserApplicationService


class Container(containers.DeclarativeContainer):
    config = providers.Configuration()

    # Factoryは呼び出しごとに新しいインスタンスを作る
    user_repository = providers.Factory(
        UserRepository,
        store_path=config.store_path
    )

    user_service = providers.Factory(
        UserService,
        user_repository=user_repository
    )

    user_application = providers.Factory(
        UserApplicationService,
        repository=user_repository,
        service=user_service
    )


if __name__ == "__main__":
    config = {
        "store_path": "store.json"
    }

    container = Container()
    container.config.from_dict(config)
    container.init_resources()

    # リポジトリ
    repository = container.user_repository()

    # アプリケーション
    app = container.user_application()

    # 初期化
    repository.clear()

    # 追加
    app.register(1, "kta", "mido")
    app.register(2, "foo", "bar")
    app.register(3, "hoge", "piyo")

    # 取得
    print(app.get(1))  # id=1 user_name=UserName(first_name='kta', last_name='mido')

    # 更新
    app.update(1, "keita", "midorikawa")

    # 削除
    app.delete(3)

■ ソフトウェアシステムを組み立てる

コントローラ

コントローラの責務は入力の変換です。
MVCにおいても、コントローラはユーザーからの入力をモデルが要求するメッセージに変換しモデルに伝えることが責務となります。

もし、コントローラがそれ以上のことをこなしているようなら、ドメインの重要なロジックがコントローラに漏れ出している可能性を疑うべきです。

■ ファクトリ

オブジェクトの生成手順が複雑な場合に、その手順をまとめるための仕組みです。

実装例

Userインスタンスを作るときのIDの採番をFactoryに委ねる仕様を実装してみます。

factory.py

import abc
from value_object import UserName
from entity import User
from repository import IUserRepository, UserRepository
from domein_service import UserService

class IUserFactory(metaclass=abc.ABCMeta):
    """ファクトリーのインターフェース"""
    @abc.abstractmethod
    def create(self, user_name: UserName) -> User:
        raise NotImplementedError

class UserFactory(IUserFactory):
    """ファクトリー"""

    def __init__(self):
        pass

    def create(self, user_name: UserName) -> User:
        id = 1  # DBに接続してインクリメントされたIDを撮ってくる処理
        return User(id=id, user_name=user_name)


class UserApplicationService:
    def __init__(
        self,
        repository: IUserRepository,
        service: UserService, 
        factory: IUserFactory
    ):
        self.service = service
        self.repository = repository
        self.factory = factory

    def register(self, first_name: str, last_name: str):
        user_name = UserName(first_name=first_name, last_name=last_name)
        user = self.factory.create(user_name)
        if self.service.exists(user):
            raise Exception(f"{user.user_name} はすでに存在しています")
        self.repository.save(user)

if __name__ == "__main__":
    store_path = "store.json"

    # リポジトリ
    repository = UserRepository(store_path)

    # ドメインサービス
    service = UserService(repository)

    # ファクトリー
    factory = UserFactory()

    # 初期化
    repository.clear()

    # アプリケーション
    app = UserApplicationService(repository, service, factory)

    # 追加
    app.register("kta", "mido")

■ ドメインのルールを守る「集約」

オブジェクト内部のインスタンス変数は直接操作・参照するのではなく、メソッドを介して操作・参照しましょう。
外部からのインスタンス変数への直接アクセスは「ドメインのルール」を散在させることになります。

結果整合性とは : あるタイミングにおいて矛盾が発生することを許容すること。

デメテルの法則 (最小知識の原則)

オブジェクトが自分以外の構造やプロパティに持っている仮定を最小限にすべきであるという原則。
つまり、オブジェクトAは別のオブジェクトBのメソッドを呼び出しても良いが、オブジェクトAはオブジェクトBのメンバであるオブジェクトCのメソッドを直接呼び出してはならないということです。

オブジェクトAがオブジェクトB自身の内部構造以上の知識を要求してしまっていることになるため

<> ルール

オブジェクトO上のメソッドMが呼び出してもいいメソッドは、以下のオブジェクトに属するメソッドに限定されます。

Oそれ自身のメソッド
Mの引数に渡されたオブジェクトのメソッド
Mの内部でインスタンス化されたオブジェクトのメソッド
Oのインスタンス変数のメソッド

<> 実装例

複数のユーザーを追加できるCircle(大学のサークル的な意味)というクラスで「デメテルの法則」のルールを考えてみます。

circle.py

from typing import List
from pydantic import BaseModel

class User(BaseModel):
    id: int
    user_name: str

    def greet(self):
        return f"hello. {self.user_name}"


class Circle(BaseModel):
    id: int = 0
    members: List[User] = []

    def is_full(self) -> bool:
        return len(self.members) >= 30

    def join(self, user: User):
        # ルール2. 引数で渡されたオブジェクトのメソッドへのアクセス
        print(user.greet())
        # ルール1. Circle自身のメソッドへのアクセス
        if self.is_full():
            raise Exception()
        # ルール4. Circleのインスタンス変数のメソッドへのアクセス
        self.members.append(user)

def main():
    circle = Circle()
    user1 = User(id=1, user_name="foo")
    # ルール3. 直接インスタンス化されたオブジェクトのメソッドへのアクセス
    # NOTE: circle.member.append(user1) のように内部のオブジェクトを直接操作するのはNG
    circle.join(user1)

if __name__ == "__main__":
    main()

■ アーキテクチャ

アーキテクチャは「方針」の意味。
何がどこに記述されるべきかといった疑問に対する解凍を明確にし、ロジックが無秩序に点在することを防ぎます。
開発者はアーキテクチャが示す方針に従うことで、「何をどこに書くのか」に振り回されないようになります。

DDDと一緒に語られることが多いアーキテクチャ

レイヤードアーキテクチャ
ヘキサゴナルアーキテクチャ
クリーンアーキテクチャ

レイヤードアーキテクチャ

プレゼンテーション層(ユーザーインターフェース)
WebインターフェースやCLIインターフェースといったようなユーザーインターフェースとアプリケーションを紐付ける。
アプリケーション層
ドメイン層の住人を取りまとめる層。
この層にはアプリケーションサービスが含まれ、ドメインオブジェクトの直接のクライアントとなり、ユースケースを実現する。
ドメイン層
ソフトウェアを適用しようとしている領域で問題解決に必要な知識を表現する。
この双に本来所属すべきドメインオブジェクトの隔離を促し、他の層へ流出しないようにする。
インフラストラクチャ層
入出力詭弁へのアクセスを提供する。
メッセージ送信や、永続化を行うモジュール(ORM, リポジトリ)などが含まれる。

ヘキサゴナルアーキテクチャ

コンセプトはアプリケーションとそれ以外のインターフェースや保存媒体はつけ外しできるようにするというものです。
ヘキサゴナルアーキテクチャはアダプタがポートの形状に合えば動作することに見立てて、ポートアンドアダプタと呼ばれることもあります。
※ まあ、要するに入出力の口をインターフェースにしましょうということ。

プライマリポート・プライマリアダプタ
入力を受け持つポートとアダプタ (CLIやWebインターフェース)
セカンダリポート・セカンダリアダプタ
外部とインタラクトするポートとアダプタ (リポジトリなど)

クリーンアーキテクチャ

クリーンアーキテクチャはユーザーインターフェースやデータストアなどの「詳細(下位コンポーネント)」(問題解決の本質と関係ないもの)を端においやり、依存を内側に向けることで、「詳細(下位コンポーネント)」が「抽象(上位コンポーネント)」に依存するという依存関係逆転の法則を達成するためのアーキテクチャです。
クリーンアーキテクチャとヘキサゴナルアーキテクチャの違いは実装仕様が詳細に言及されているか否か。(ヘキサゴナルアーキテクチャはポートとアダプタにより付け外し可能にするという方針だけがある)

クリーンアーキテクチャのEntityはDDDのエンティティとは異なり、ビジネスルールをカプセル化したオブジェクトを意味します。(ドメインオブジェクトに近い概念)

clean_architecture.py

import abc
from pydantic import BaseModel
from repository import UserRepository, IUserRepository

class UserData(BaseModel):
    id: int
    user_name: str

class UserUpdateOutputData(BaseModel):
    """出力データ"""
    user_data: UserData

class UserGetInputData(BaseModel):
    """入力データ"""
    id: int


#
# [Use Case Output Port]
#
class IUserGetOutputPort(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def output(self, output_data: UserUpdateOutputData):
        raise NotImplementedError

#
# [Presenter]
#
class UserGetPresenter(IUserGetOutputPort):
    def output(self, output_data: UserUpdateOutputData):
        print(output_data)

#
# [Use Case Input Port]
#
class IUserGetInputPort(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def handle(self, input_data: UserGetInputData):
        raise NotImplementedError

#
# [Use Case Interactor]
#
class UserGetInteractor(IUserGetInputPort):
    """アプリケーションサービスのgetメソッドをそのままクラスにしたもの
    アプリケーションサービスと異なる点は、結果の出力先がpresenterオブジェクトであること
    """
    def __init__(self, repository: IUserRepository, presenter: IUserGetOutputPort):
        self.repository = repository
        self.presenter = presenter

    def handle(self, input_data: UserGetInputData):
        user = self.repository.find(input_data.id)
        if user is None:
            return
        user_data = UserData(id=user.id, user_name=str(user.user_name))
        output_data = UserUpdateOutputData(user_data=user_data)
        self.presenter.output(output_data)

#
# [Controller]
#
def controller(interactor: IUserGetInputPort):
    input_data = UserGetInputData(id=1)
    interactor.handle(input_data)

if __name__ == "__main__":
    repository = UserRepository("store.json")
    presenter = UserGetPresenter()
    interactor = UserGetInteractor(repository, presenter)
    controller(interactor)