概要
- DB操作クラス周りの設計実装における結論
ありがちな悩み
- 各テーブルに対してCRUD操作するクラスを書きたい
- テーブルごとにクラスを分割するとconnectionやsessionの取り回しが悪い
- かといって単一クラスでやると肥大化するからファイル分割したい
Mix-in とは
- オブジェクト指向設計における多重継承を利用した設計手法
- コアとなるクラスに対し、Mix-inクラスを多重継承することで機能を追加する
class FooBarCore:
def __init__(self):
self.connection = FooBarConnection()
class ReadMixin:
def read(self):
return self.connection.read()
class WriteMixin:
def write(self):
self.connection.write()
class FooBarRW(FooBarCore, ReadMixin, WriteMixin):
pass
ディレクトリ構成
- ORMクラスごとにmixinモジュールを用意するイメージ
[project-root]
+-- src
+-- models
+-- db
+-- mixin
+-- base.py
+-- foo.py
+-- bar.py
+-- foo_bar_mapping.py
+-- orm
+-- base.py
+-- foo.py
+-- bar.py
+-- foo_bar_mapping.py
+-- core.py
+-- crud.py
+-- errors.py
データベース操作クラス作成
こちらの記事を参照
https://qiita.com/yuki-alost/items/4d24178a85e1d9a787b2
- クラス名を
DatabaseCoreにしておく
core.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
class DatabaseCore:
def __init__(self, url: str):
self.__engine = create_engine(
# 以下略
ベースクラス定義
- 各Mix-inクラスでの共通処理などを書いておく
- 例としてユーザを一意に特定する処理を実装した
- この例では
user_idではなくuser_nameで外部とやり取りする設計思想
- この例では
mixin/base.py
from sqlalchemy.exc import NoResultFound
from models.db.errors import UserNotFoundException
from models.db.orm.user import User
class MixinBase:
def query_user(self, session, user_name: str) -> User:
try:
return session.query(User).filter(User.user_name == user_name).one()
except NoResultFound as e:
raise UserNotFoundException from e
Mix-inクラス定義… の前に
- ORMクラスに
to_dict()メソッドを用意しておく - dict化したレコードはfrontendに渡すことも考慮してキャメルケースにするべき
orm/base.py
from sqlalchemy.orm import declarative_base
Base = declarative_base()
def __to_camel_case(key: str) -> str:
words = key.split('_')
return words[0] + ''.join(x.capitalize() for x in words[1:])
def to_dict(orm) -> dict:
return {
__to_camel_case(k): v
for k, v in vars(orm).items()
if not k.startswith('_sa')
}
orm/user.py
from .base import Base, to_dict
class User(Base):
__tablename__ = 'user'
# カラム省略
def to_dict(self) -> dict:
return to_dict(self)
Mix-inクラス定義
- 対象のORMクラスが定義するテーブルに対するCRUD操作を実装する
-
xxx_in_session()としてセッション内の処理をメソッド分割している理由は後述
mixin/user.py
import uuid
from models.db.orm.user import User
from .base import MixinBase
class UserCreateMixin(MixinBase):
def create_user(
self,
*,
user_name: str,
user_label: str = '',
):
with self.create_session() as session:
try:
self.create_user_in_session(
session,
user_name=user_name,
user_label=user_label,
)
session.commit()
except Exception:
session.rollback()
raise
def create_user_in_session(
self,
session,
*,
user_name: str,
user_label: str = '',
):
user = User(
user_id=str(uuid.uuid4()).replace('-', ''),
user_name=user_name,
user_label=user_label if user_label else user_name,
)
session.add(user)
session.flush()
class UserReadMixin(MixinBase):
def read_user(self, user_name: str) -> dict:
with self.create_session() as session:
return self.read_user_in_session(session, user_name)
def read_user_all(self) -> list[dict]:
with self.create_session() as session:
return self.read_user_all_in_session(session)
def read_user_in_session(self, session, user_name: str) -> dict:
return self.query_user(session, user_name).to_dict()
def read_user_all_in_session(self, session) -> list[dict]:
return [ r.to_dict() for r in session.query(User).all() ]
class UserUpdateMixin(MixinBase):
def update_user(
self,
*,
target_user_name: str,
user_name: str = '',
user_label: str = '',
):
with self.create_session() as session:
try:
self.update_user_in_session(
session,
target_user_name=target_user_name,
user_name=user_name,
user_label=user_label,
)
session.commit()
except Exception:
session.rollback()
raise
def update_user_in_session(
self,
session,
*,
target_user_name: str,
user_name: str = '',
user_label: str = '',
):
user = self.query_user(session, target_user_name)
if user_name:
user.user_name = user_name
if user_label:
user.user_label = user_label
session.flush()
class UserDeleteMixin(MixinBase):
def delete_user(self, user_name: str):
with self.create_session() as session:
try:
self.delete_user_in_session(session, user_name)
session.commit()
except Exception:
session.rollback()
raise
def delete_user_in_session(self, session, user_name: str):
session.query(User).filter(User.user_name == user_name).delete()
session.flush()
CRUD実現クラス作成
- 必要な機能をMix-in継承で取り込む
crud.py
from models.db.core import DatabaseCore
from models.db.mixin.group import GroupReadMixin
from models.db.mixin.user import (
UserCreateMixin,
UserReadMixin,
UserUpdateMixin,
UserDeleteMixin,
)
from models.db.mixin.user_group_mapping import (
UserGroupMappingCreateMixin,
UserGroupMappingReadMixin,
UserGroupMappingDeleteMixin,
)
class DatabaseManager(
DatabaseCore,
GroupReadMixin,
UserCreateMixin,
UserReadMixin,
UserUpdateMixin,
UserDeleteMixin,
UserGroupMappingCreateMixin,
UserGroupMappingReadMixin,
UserGroupMappingDeleteMixin,
):
pass
CRUD実現クラス拡張
- 以下はユーザ作成とグループへのマッピングを1つのセッション内で完結させる例
-
xxx_in_session系メソッドを定義しておいた理由がこれ - 複数のテーブルに干渉する処理はこの形式で実装するのが良い
crud.py
class DatabaseManager(
# 省略
):
def create_user_with_group_mapping(
self,
*,
user_name: str,
user_label: str = '',
group_name: str,
):
with self.create_session() as session:
try:
self.create_user_in_session(
session,
user_name=user_name,
user_label=user_label,
)
self.create_user_group_mapping_in_session(
session,
user_name=user_name,
group_name=group_name,
)
session.commit()
except Exception:
session.rollback()
raise