概要

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