ボトムアップドメイン駆動設計の**ユーザに関連する処理の「登録・変更・削除・取得」**することについて、Pythonで書いてみた。

C#で書かれているコードをPythonに翻訳したので、厳密に再現できていない箇所もあります。また、それぞれの呼び出しをテストコードで表し、ソースコードレベルで目的を明確にしました。

筆者が訳した「pythonでボトムアップDDD」
PythonでボトムアップDDD 【値オブジェクト】
PythonでボトムアップDDD 【エンティティ】
PythonでボトムアップDDD 【ドメインサービス】
PythonでボトムアップDDD【値オブジェクト・エンティティ・ドメインサービスを利用する】
PythonでボトムアップDDD 【リポジトリ】
PythonでボトムアップDDD 【テスト用のリポジトリ】
PythonでボトムアップDDD 【ユーザの登録・変更・削除・取得など】

バージョン

Python 3.7.0

目次

ユーザ関連の処理を実装
テスト用のリポジトリを実装
テストを実装
参考文献
おまけ

ディレクトリ構成

ファイルを分割しているので、ディレクトリ構成を載せておきます。

DDD_application_service
├── test
│   └── test_user_application_service.py
├── user.py
├── user_application_service.py
├── user_repository.py
├── user_service.py
└── user_summary_model.py

ユーザ関連の処理を実装

user_application_service.py

import uuid
from typing import List

from DDD_application_service.user import User, FullName, Username, UserId
from DDD_application_service.user_repository import IUserRepository
from DDD_application_service.user_service import UserService
from DDD_application_service.user_summary_model import UserSummaryModel


# このクラスに以下のメソッドを追加していきます。
class UserApplicationService:
    def __init__(self, user_repository: IUserRepository):
        self.user_repository = user_repository
        self.user_service = UserService(user_repository)

ユーザの追加

user_application_service.py

def register_user(self, username: str, first_name: str, family_name: str) -> None:
    user = User(UserId(str(uuid.uuid4())), Username(username), FullName(first_name, family_name))

    if self.user_service.is_duplicated(user):
        raise ValueError("重複しています")
    else:
        self.user_repository.save(user)

ユーザ情報の変更

user_application_service.py

def change_user_info(self, user_id: str, username: str, first_name: str, family_name: str) -> None:
    target_id = UserId(user_id)
    target = self.user_repository.find_by_user_id(target_id)

    if target == None:
        raise Exception("not found. target id:" + user_id)

    new_username = Username(username)
    target.change_username(new_username)

    new_full_name = FullName(first_name, family_name)
    target.change_name(new_full_name)

    self.user_repository.save(target)

ユーザの削除

user_application_service.py

def remove_user(self, user_id: str) -> None:
    target_id = UserId(user_id)
    target = self.user_repository.find_by_user_id(target_id)

    if target == None:
        raise Exception("not found. target id:" + user_id)

    self.user_repository.remove(target)

ユーザを取得

user_application_service.py

def find_all(self) -> List[UserSummaryModel]:
    users = self.user_repository.find_all()

    return [UserSummaryModel(user) for user in users]

ユーザをまとめて取得する際に、必要な情報のみを返すために、それようのオブジェクトを作ります。

user_summary_model.py

from DDD_application_service.user import User


class UserSummaryModel:
    def __init__(self, user: User):
        self.id = user.id.value
        self.first_name = user.name.first_name

テスト用のリポジトリを実装

テスト用のリポジトリとして、ユーザ情報を配列に格納しておきます。

user_repository.py

import uuid
from abc import ABCMeta, abstractmethod
from typing import Union, List

from DDD_application_service.user import UserId, Username, User, FullName


class IUserRepository(metaclass=ABCMeta):
    @abstractmethod
    def find_by_user_id(self, user_id: UserId) -> Union[User, None]:
        pass

    @abstractmethod
    def find_by_username(self, username: Username) -> Union[User, None]:
        pass

    @abstractmethod
    def find_all(self) -> Union[List[User], None]:
        pass

    @abstractmethod
    def save(self, user: User) -> None:
        pass

    @abstractmethod
    def remove(self, user: User) -> None:
        pass


class InMemoryUserRepository(IUserRepository):
    def __init__(self):
        self.data = []

        user_id_1 = UserId(str(uuid.uuid4()))
        initial_user = User(user_id_1, Username("kota"), FullName("こうた", "まつおか"))
        self.data.append(initial_user)

        user_id_2 = UserId(str(uuid.uuid4()))
        initial_user_2 = User(user_id_2, Username("daiki"), FullName("だいき", "おおた"))
        self.data.append(initial_user_2)

    def find_by_user_id(self, user_id: UserId) -> Union[User, None]:
        target_user = [user for user in self.data if user.id.value == user_id.value]
        if target_user:
            return target_user[0]
        else:
            return None

    def find_by_username(self, username: Username) -> Union[User, None]:
        target_user = [user for user in self.data if user.username.value == username.value]
        if target_user:
            return target_user[0]
        else:
            return None

    def find_all(self) -> List[User]:
        return self.data

    def save(self, user: User) -> None:
        self.data.append(user)

    def remove(self, user: User) -> None:
        self.data.remove(user)

テストを実装

「呼び出し方を明らかにするため」と「動くコードであることを担保するため」にテストも書きます。このテストたちがあるおかげで、リファクタリングをしやすいですし、設計の見直しも容易にできます。

test/test_user_application_service.py

import unittest

from DDD_application_service.user import Username, UserId
from DDD_application_service.user_application_service import UserApplicationService
from DDD_application_service.user_repository import InMemoryUserRepository


class TestUserApplicationService(unittest.TestCase):
    def setUp(self):
        self.in_memory_user_repository = InMemoryUserRepository()
        self.user_application_service = UserApplicationService(self.in_memory_user_repository)

    def test_ユーザを登録する(self):
        username = "yuki"

        self.user_application_service.register_user(username, "ゆうき", "まつい")
        actual = self.in_memory_user_repository.find_by_username(Username(username))

        self.assertEqual(username, actual.username.value)

    def test_ユーザを登録する_usernameが重複時は保存できない(self):
        with self.assertRaises(ValueError):
            self.user_application_service.register_user("kota", "こうた", "まつい")

    def test_ユーザ情報を変更する(self):
        new_username = "kotaro"
        new_first_name = "こたろう"
        new_family_name = "まつお"
        target = self.in_memory_user_repository.find_by_username(Username("kota"))
        target_id = target.id.value

        self.user_application_service.change_user_info(target_id, new_username, new_first_name, new_family_name)

        actual = self.in_memory_user_repository.find_by_user_id(UserId(target_id))

        self.assertEqual(new_username, actual.username.value)
        self.assertEqual(new_first_name, actual.name.first_name)
        self.assertEqual(new_family_name, actual.name.family_name)

    def test_ユーザを削除する(self):
        target = self.in_memory_user_repository.find_by_username(Username("kota"))
        target_id = target.id.value
        self.user_application_service.remove_user(target_id)

        actual = self.in_memory_user_repository.find_by_user_id(UserId(target_id))

        self.assertIsNone(actual)

    def test_ユーザ一覧を取得する(self):
        actual = self.user_application_service.find_all()

        target_1 = self.in_memory_user_repository.find_by_username(Username("kota"))
        target_2 = self.in_memory_user_repository.find_by_username(Username("daiki"))

        self.assertEqual(target_1.id.value, actual[0].id)
        self.assertEqual(target_2.id.value, actual[1].id)


if __name__ == "__main__":
    unittest.main()

参考文献

ボトムアップドメイン駆動設計

おまけ

参考文献に載せたブログ記事にはいつも助けられています。ぜひ、みなさんも読みながら、ご自身が使い慣れている言語で写経してみてください。

Renttleというサービスを開発中です。ぜひ、使ってみて、レビューをください。