Edited at

PythonでボトムアップDDD 【ユーザの登録・変更・削除・取得など】

ボトムアップドメイン駆動設計ユーザに関連する処理の「登録・変更・削除・取得」することについて、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というサービスを開発中です。ぜひ、使ってみて、レビューをください。