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_test_repository

├── in_memory_user_reposiotry.py
├── programn.py
├── test
│   └── test_program.py
├── user.py
└── user_service.py


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

テスト用にSQLを用意するのは、面倒なことが多々あるので、配列で代替します。


user_repository.py

import uuid

from typing import Union

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

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

user_id = UserId(str(uuid.uuid4()))
initial_user = User(user_id, Username("kota"), FullName("こうた", "まつい"))
self.data.append({user_id: initial_user})

def find(self, username: Username) -> Union[User, None]:
# もう少しきれいに書けるはず
target_user = [user for user in self.data if list(user.values())[0].username.value == username.value]
if target_user:
return list(target_user[0].values())[0]
else:
return None

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



リポジトリの依存を除去


リポジトリのインターフェースの用意する

インターフェイス用の完全抽象クラスを作ります。


user_repository.py

import uuid

from abc import ABCMeta, abstractmethod
from typing import Union

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

class IUserRepository(metaclass=ABCMeta):
@abstractmethod
def find(self, username: Username) -> Union[User, None]:
pass

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

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

user_id = UserId(str(uuid.uuid4()))
initial_user = User(user_id, Username("kota"), FullName("こうた", "まつい"))
self.data.append({user_id: initial_user})

def find(self, username: Username) -> Union[User, None]:
# もう少しきれいに書けるはず
target_user = [user for user in self.data if list(user.values())[0].username.value == username.value]
if target_user:
return list(target_user[0].values())[0]
else:
return None

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



リポジトリを外部から渡す

ここからの実装は、こちらの記事で実装した内容を参考にしてください。


test/test_program.py

import unittest

from DDD_test_repository.in_memory_user_reposiotry import InMemoryUserRepository
from DDD_test_repository.programn import Program
from DDD_test_repository.user import Username

class TestProgram(unittest.TestCase):
def test_usernameが重複時は保存できない(self):
in_memory_user_repository = InMemoryUserRepository()
with self.assertRaises(ValueError):
Program().creat_user("kota", "こうた", "まつい", in_memory_user_repository)

def test_usernameが重複していない時は保存してそのusernameが一件だけ存在するか(self):
username = "yuki"
in_memory_user_repository = InMemoryUserRepository()

Program().creat_user(username, "ゆうき", "まつい", in_memory_user_repository)
actual = in_memory_user_repository.find(Username(username))

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

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



user_service.py

from typing import Union

from DDD_test_repository.in_memory_user_reposiotry import IUserRepository
from DDD_test_repository.user import User

class UserService:
def __init__(self, user_repository: IUserRepository):
self.user_repository = user_repository

def is_duplicated(self, user: User) -> Union[User, None]:
return self.user_repository.find(user.username)



program.py

import uuid

from DDD_test_repository.in_memory_user_reposiotry import IUserRepository
from DDD_test_repository.user import User, UserId, Username, FullName
from DDD_test_repository.user_service import UserService

class Program:
def creat_user(self, username: str, fist_name:str, family_name: str, user_repository: IUserRepository):
user_repository = user_repository
user = User(UserId(str(uuid.uuid4())),
Username(username),
FullName(fist_name, family_name))

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



参考文献


おまけ

これでテスト用のリポジトリに取り替えが、できるようになりました。コードとの結合が薄いテストが実装できることは嬉しいことですね。