ボトムアップドメイン駆動設計のリポジトリについて、Pythonで書いてみた。
C#で書かれているコードをPythonに翻訳したので、厳密に再現できていない箇所もあります。また、それぞれの呼び出しをテストコードで表し、ソースコードレベルで目的を明確にしました。
用語の説明やコードの詳細説明に関して、下記の参考文献を参照ください。
参考文献: ボトムアップドメイン駆動設計
バージョン
- Python 3.7.0
目次
- リポジトリの実装
- DBに関する処理を除去
- テスト
- 参考文献
- おまけ
ディレクトリ構成
ファイルを分割しているので、ディレクトリ構成を載せておきます。
DDD_repository
├── program.py
├── test
│ └── test_program.py
├── user.py
├── user_repository.py
└── user_service.py
リポジトリの実装
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と取り替えが容易にできることを想定して、実装を加えていきます。