PythonでボトムアップDDD 【リポジトリ】 #Python

ボトムアップドメイン駆動設計のリポジトリについて、Pythonで書いてみた。

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

用語の説明やコードの詳細説明に関して、下記の参考文献を参照ください。

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

バージョン

Python 3.7.0

リポジトリの実装

DBに関する処理はすべて、リポジトリ内に押し込みます。

user_repository.py

import sqlite3
from typing import Union

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


class UserRepository:
    def find(self, username: Username) -> Union[User, None]:
        conn = sqlite3.connect('sample.db')
        c = conn.cursor()
        t = (username.value,)
        c.execute("SELECT * FROM users WHERE username=?", t)
        user_tuple = c.fetchone()

        if user_tuple:
            # indexで取得しているのは改良の余地あり
            id = UserId(user_tuple[0])
            username = Username(user_tuple[1])
            fullname = FullName(user_tuple[2], user_tuple[3])

            return User(id, username, fullname)
        else:
            return None

    def save(self, user: User):
        conn = sqlite3.connect('sample.db')
        c = conn.cursor()
        t = (user.id.value, user.username.value, user.name.first_name, user.name.family_name)
        c.execute("INSERT INTO users VALUES (?,?,?,?)", t)
        conn.commit()
        conn.close()

DBに関する処理を除去

DBに関する処理はすべてリポジトリに押し込むため、エンティティやドメインサービスから、それらの処理を除去します。

モデル

user.py

from __future__ import annotations

from dataclasses import dataclass


@dataclass(frozen=True)
class UserId:
    value: str

@dataclass(frozen=True)
class Username:
    value: str

@dataclass(frozen=True)
class FullName:
    first_name: str
    family_name: str

@dataclass
class User:
    id: UserId
    username: Username
    name: FullName

    def change_username(self, new_username: Username):
        self.username = new_username

    def change_name(self, new_name: FullName):
        self.name = new_name

    def __eq__(self, other: User):
        # idのみで比較する
        return isinstance(other, User) and (self.id == other.id)

ドメインサービス

ドメインサービスからも、すっきりDBに関する記述は削除できる。

user_service.py

from typing import Union

from DDD_repository.user import User
from DDD_repository.user_repository import UserRepository


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

ユーザ登録のロジック

program.py

import uuid

from DDD_repository.user import User, UserId, Username, FullName
from DDD_repository.user_repository import UserRepository
from DDD_repository.user_service import UserService


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

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

テスト

前回のテストに加えて、usernamneでUserを検索できるかのテストを追加した。

test/test_program.py

import sqlite3
import unittest
import uuid

from DDD_repository.program import Program
from DDD_repository.user import Username
from DDD_repository.user_repository import UserRepository


class TestProgram(unittest.TestCase):
    def setUp(self):
        conn = sqlite3.connect("sample.db")
        c = conn.cursor()
        c.execute("""CREATE TABLE users(id text, username text, first_name text, family_name text)""")
        users = [(str(uuid.uuid4()), "kota", "こうた", "まつおか"), (str(uuid.uuid4()), "kazuo", "かずお", "まつい")]
        c.executemany("INSERT INTO users VALUES (?,?,?,?)", users)
        conn.commit()
        conn.close()

    def tearDown(self):
        conn = sqlite3.connect("sample.db")
        c = conn.cursor()
        c.execute("""DROP TABLE users""")
        conn.commit()
        conn.close()

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

    def test_usernameが重複していない時は保存してそのusernameが一件だけ存在するか(self):
        username = "yuki"
        Program().creat_user(username, "ゆうき", "まつい")

        conn = sqlite3.connect('sample.db')
        c = conn.cursor()
        t = (username,)
        c.execute("SELECT * FROM users WHERE username=?", t)

        self.assertEqual(1, c.fetchall().__len__())

    def test_usernameでUserを検索する(self):
        username = Username("kota")

        user_repository = UserRepository()
        actual = user_repository.find(username)

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

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

参考文献

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

おまけ

DBに関する処理が、リポジトリに固まると、ビジネスロジック部分のコードがすっきりします。
これ移行の記事で、テスト用のDBと取り替えが容易にできることを想定して、実装を加えていきます。

PythonでボトムアップDDD 【リポジトリ】