PythonでのDBSessionクラス

SQLAlchemy とは

SQLAlchemy とは、python の中では最もよく利用されている ORM の一つです。
ORM 以外にも以下の機能を持ちます。

1.データベースへの接続、SQL の実行 2.メタデータ
3.SQL Expression Language
4.ORM

SQLAlchemy は様々なデータベースに対して接続して SQL を実行することができます。
サポートする DBMS は以下のとおりです。有名どころは大抵利用可能です。

1.Firebird
2.Microsoft SQL Server
3.MySQL
4.Oracle
5.PostgreSQL
6.SQLite
7.Sybase

環境

Windows 10
python 3.8.10

SQLAlchemy のインストール

pip install sqlalchemy

実装

from sqlalchemy import create_engine, event as sqlalchemy_event
from sqlalchemy.orm import sessionmaker, make_transient

def db_engine():
    """DBセッション生成するクラスを取得する(非公開)
    データベースエンジンの設定情報を取得する
    Returns:
        データベースエンジンの設定情報
    """
    # データベース名はSecretsManagerから取得前提
    return create_engine(
        f'postgresql://{username}:{password}@{host}:{port}/{dbname}',
        encoding='utf-8',
        echo=LOG_LEVEL == 'DEBUG',  # ログレベルがDEBUGの場合、発行するSQLをログ出力する
    )

def _session_cls():
    """DBセッション生成するクラスを取得する(非公開)
    データベースエンジンの設定情報を取得する
    Returns:
        データベースエンジンの設定情報
    """
    return sessionmaker(
        autocommit=False,
        autoflush=True,
        bind=db_engine(),
    )

class DBSession:
    """DB接続クラス
    使い方)正常終了時はCommit,例外発生時はRollbackを実施する
        with DBSession() as new_session:
            new_session.query()....

    使い方)Modelを定義する場合(検索)
        with DBSession() as new_session:
            for model_object in new_session.query(ModelName).order_by(ModelName.field_name):
                xxxx(model_object)

    使い方)生のSQLを利用する場合(検索)
        from sqlalchemy import text
        stmt = text('SELECT * FROM TableName WHERE xxxx = :name ORDER BY field_name').columns(yyyy, zzzz)
        with DBSession() as new_session:
            for model_object in new_session.query(ModelName).from_statement(stmt).params(name='hoge').all():
                xxxx(model_object)
    Args:
        for_update: 検索したオブジェクトに対して更新（削除）を実施するかどうかのフラグ（デフォルト：False）
    Attributes:
        session:DBセッション
    """

    def __init__(self, for_update=False):
        self.session = _session_cls()()

        if not for_update:
            # セッションにオブジェクトロード時（検索）のイベント処理を追加
            @sqlalchemy_event.listens_for(self.session, 'loaded_as_persistent')
            def _object_make_transient(session, obj):
                # オブジェクトの永続性を初期化する
                make_transient(obj)

    def __enter__(self):
        return self.session

    def __exit__(self, exc_type, exc_val, exc_tb):

        if exc_type is None and exc_val is None and exc_tb is None:
            self.commit()
        else:
            self.rollback()

    def commit(self):
        self.session.commit()

    def rollback(self):
        self.session.rollback()