Skip to content

Instantly share code, notes, and snippets.

@mohira

mohira/bu_ddd_repository.py

Last active December 4, 2018 17:09
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save mohira/10b4d3aae0115cf0e573e0472a8ad8a1 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save mohira/10b4d3aae0115cf0e573e0472a8ad8a1 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
bu_ddd_repository.py
"""
ボトムアップドメイン駆動設計のリポジトリにフォーカスしたPython実装例
参考: [ボトムアップドメイン駆動設計](https://nrslib.com/bottomup-ddd/#outline__4)
## Version
Python 3.7.0
## 参考記事との違いなど
### Userエンティティではない
- このスクリプトでは単純なUserクラスにしている
### DBはSQLite3
- 環境構築が簡単なのでSQLite3を選択
- sqlite3モジュールはRAM上にデータベースをつくれるが、その機能は使っていない
- 今回は`InMemoryUserRepository`との比較がメインであるため
"""
import unittest
from abc import ABCMeta, abstractmethod
from dataclasses import dataclass
from db import UseDatabaseAPI
@dataclass
class User:
name: str
age: int
class IUserRepository(metaclass=ABCMeta):
@abstractmethod
def find(self, name: str) -> User:
pass
@abstractmethod
def save(self, user: User) -> None:
pass
class UserRepository(IUserRepository):
def __init__(self):
self.db_api = UseDatabaseAPI("sample.db")
def find(self, name: str) -> User:
with self.db_api as cursor:
sql = "SELECT * FROM t_user WHERE name = ?"
row = cursor.execute(sql, (name,)).fetchone()
if row:
return User(name=row[0], age=row[1])
def save(self, user: User) -> None:
with self.db_api as cursor:
sql = "INSERT INTO t_user (name, age) VALUES (?, ?)"
cursor.execute(sql, (user.name, user.age))
class InMemoryUserRepository(IUserRepository):
def __init__(self):
self.data = []
def find(self, name: str) -> User:
for user in self.data:
if user.name == name:
return user
def save(self, user: User) -> None:
self.data.append(user)
class UserService:
def __init__(self, user_repository: IUserRepository):
self.user_repository = user_repository
def is_duplicated(self, user: User) -> bool:
searched = self.user_repository.find(user.name)
return searched is not None
class Program:
def create_user(self, name: str, age: int, user_repository: IUserRepository):
user = User(name, age)
user_service = UserService(user_repository)
if user_service.is_duplicated(user):
raise ValueError("ユーザー名が重複しています")
user_repository.save(user)
class TestProgram(unittest.TestCase):
def test_ユーザー名の重複は許可しない(self):
program = Program()
repository = InMemoryUserRepository()
program.create_user("Alice", 10, repository)
with self.assertRaises(ValueError):
program.create_user("Alice", 99, repository)
if __name__ == "__main__":
unittest.main()
Raw
db.py
import sqlite3
class UseDatabaseAPI:
"""sqlite3用のデータベースコンテキストマネージャクラス
別になくても問題ないがコネクション周りの余計なコードを隠せるので採用してみた
"""
def __init__(self, database: str):
self.database = database
def __enter__(self) -> sqlite3.Cursor:
self.conn = sqlite3.connect(self.database)
self.cursor = self.conn.cursor()
return self.cursor
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
self.conn.commit()
self.cursor.close()
self.conn.close()
def init_db():
"""わかりやすくするためのデータベース初期化関数"""
init_sql = """
DROP TABLE IF EXISTS t_user;
CREATE TABLE t_user (
name TEXT,
age INTEGER
);
INSERT INTO
t_user (name, age)
VALUES
('Bob', 10),
('Tom', 20),
('Ken', 30)
; """
with UseDatabaseAPI("sample.db") as cursor:
cursor.executescript(init_sql)
if __name__ == "__main__":
init_db()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment