LoginSignup
5
3

More than 5 years have passed since last update.

kota matsuokaの1人アドベントカレンダー ~Pythonで0からサービスを開発~ Advent Calendar 2018Day 15
@kotmats(Kot Mats)

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

Last updated at Posted at 2018-12-15

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

5
3
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
5
3