Pythonでデータベースを扱うときに何をするべきか

概要

Pythonで書いたアプリからDBを使いたいけどどうするの、という方向け

方針

• DBを初期化・操作する際にSQLを直接実行しない
  • SQLインジェクション耐性のため
  • ORMと専用のライブラリであるSQLAlchemyを最大限活用する

必要モジュールのインストール

以下を実行

• pipenv install sqlalchemy
• pipenv install sqlalchemy-utils

追加で使用するRDBMSに対応するモジュールも導入する
今回はPostgreSQLであれば以下の通り

• pipenv install psycopg[binary]

  [binary] を指定することで libpq などの別途導入が不要になる

テーブル設計

こちらの記事を参照
https://qiita.com/yuki-alost/items/644fe7df2e8c9be7376a

ディレクトリ構成

  [project-root]
       +-- src
             +-- models
                   +-- db
                         +-- orm
                               +-- base.py
                               +-- foo.py
                               +-- bar.py
                               +-- foo_bar_mapping.py
                         +-- core.py

メタクラス定義

以下のように定義した Base を以降作成するORMクラスの親クラスとする

base.py

from sqlalchemy.orm import declarative_base
Base = declarative_base()

ORMクラス定義

エンティティテーブル

• 外部キーを持たないテーブルの例

from sqlalchemy import Column
from sqlalchemy.sql.functions import current_timestamp
from sqlalchemy.types import CHAR, DateTime, String

from .base import Base


class User(Base):
    __tablename__ = 'user'

    user_id = Column(CHAR(32), primary_key=True)
    user_name = Column(String(40), unique=True, nullable=False)
    user_label = Column(String(255), nullable=False)
    create_datetime = Column(
        DateTime(timezone=True),
        nullable=False,
        default=current_timestamp(),
    )
    update_datetime = Column(
        DateTime(timezone=True),
        nullable=False,
        default=current_timestamp(),
        onupdate=current_timestamp(),
    )

マッピングテーブル

• 外部キーおよび複合ユニークキーを持つテーブルの例
• ORMリレーションはデフォルトでは定義しなくて良いので割愛
  • 頻繁に JOIN する想定のテーブル間には定義しておくと良い

from sqlalchemy import Column, ForeignKey
from sqlalchemy.schema import UniqueConstraint
from sqlalchemy.sql.functions import current_timestamp
from sqlalchemy.types import CHAR, DateTime, Integer

from .base import Base


class UserGroupMapping(Base):
    __tablename__ = 'user_group_mapping'
    __table_args__ = (
        UniqueConstraint('user_id', 'group_id', name='unique_user_group'),
    )
    
    mapping_id = Column(
        Integer,
        autoincrement=True,
        primary_key=True,
    )
    user_id = Column(
        CHAR(32),
        ForeignKey('user.user_id', onupdate='CASCADE', ondelete='CASCADE'),
        nullable=False,
    )
    group_id = Column(
        CHAR(32),
        ForeignKey('group.group_id', onupdate='CASCADE', ondelete='CASCADE'),
        nullable=False,
    )
    create_datetime = Column(
        DateTime(timezone=True),
        nullable=False,
        default=current_timestamp(),
    )
    update_datetime = Column(
        DateTime(timezone=True),
        nullable=False,
        default=current_timestamp(),
        onupdate=current_timestamp(),
    )

データベース操作クラス作成

• RDBMSの起動、ユーザ作成、データベースの作成等は解説を割愛
  • 適切な環境変数を与えてDockerコンテナを立ち上げれば諸々自動で作成される
• デフォルトスキーマを使用する実装になっているので必要があれば変更を
• create_engine の引数でコネクションプールのサイズやリサイクル頻度を設定する
  • 推奨値はプロジェクトの要件次第だが、今回は大規模案件を想定した値を設定

core.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


class DatabaseManager:

    def __init__(self, url: str):
        self.__engine = create_engine(
            url,
            pool_pre_ping=True,
            pool_size=100,
            pool_recycle=3600,
        )
        self.__connection = self.__engine.connect()
        self.__create_session = sessionmaker(bind=self.__engine)
    
    @property
    def engine(self):
        return self.__engine

    @property
    def connection(self):
        return self.__connection
    
    def close(self):
        if self.connection:
            self.connection.close()
        if self.engine:
            self.engine.dispose()
    
    def commit(self):
        self.connection.commit()

    def create_session(self):
        return self.SessionContext(self.__create_session())

    class SessionContext:
        def __init__(self, session):
            self.__core = session

        def __enter__(self):
            return self.__core

        def __exit__(self, ex_type, ex_value, traceback):
            self.__core.close()

データベースにテーブルを作成

• 全てのORMクラスをインポートすること

from models.db.orm.foo import Foo
from models.db.orm.bar import Bar
from models.db.orm.foo_bar_mapping import FooBarMapping

from models.db.core import DatabaseCore
from models.db.orm.base import Base


database = DatabaseManager('postgresql+psycopg://dbuser:dbpass@localhost:5432/mydb')
Base.metadata.create_all(bind=database.engine)
database.commit()