はじめに
SQLAlchemyはPythonで一般的に広く使用されているORM（Object Relational Mapping）ライブラリの1つで、データベース操作の複雑さを抽象化し、より簡潔で読みやすいコードを書くことができます。
本記事では、SQLAlcemyのデータベースセッションの管理に焦点を当て、セッションのライフサイクルをcontext managerを利用して管理する方法を記載します。
なお、Pythonでcontext managerを実装する方法として、context managerクラスを作る方法やcontextlibのcontextmanager()デコレータを使う方法がありますが、本記事では前者のクラスを作る方法を載せます。
前提
本記事では以下のUserモデルを用意します。
def generate_uuid():
return str(uuid.uuid4())
class UserModel(Base):
__tablename__ = "user"
id = Column(String, primary_key=True, default=generate_uuid)
name = Column(String, nullable=False)
email = Column(String, nullable=False, unique=True)
実装コード
context managerを使ってセッション管理を行うクラスを作ります。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
class SessionManager:
def __init__(self):
self.session_maker = sessionmaker(
bind=create_engine("DBDriver:///Path")
)
def __enter__(self):
self.session = self.session_maker()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is not None:
self.session.rollback()
self.session.close()
def commit(self):
""" SQLAlchemyのcommitメソッドのWrapper """
# コミット前になにかする場合はここに処理を記述する
self.session.commit()
def rollback(self):
""" SQLAlchemyのrollbackメソッドのWrapper """
# ロールバック前になにかする場合はここに処理を記述する
self.session.rollback()
上記のsession_managerを用いた呼び出し元（データベース操作）の実装サンプルは以下です。
from models import UserModel
from session_manager import SessionManager
def create_user(name, email):
""" 新しいユーザーをデータベースに追加 """
with SessionManager() as manager:
new_user = UserModel(name=name, email=email)
manager.session.add(new_user)
manager.commit()
return new_user
# create_user()の呼び出し例
try:
# ユーザーを作成
user = create_user("test", "test@example.com")
except Exception as e:
print(f"Error: {e}")
説明
context managerの簡単な動作も含めて実装コードの処理内容を説明していきます。
__init__()
# セッションファクトリオブジェクトを生成
def __init__(self):
self.session_maker = sessionmaker(
bind=create_engine("DBDriver:///Path")
)
context managerのインスタンス作成時に呼び出される初期化メソッドです。
上記の例では、呼び出し元で定義したwith SessionManager() as managerの実行タイミングで呼び出されます。
__init__()内でSQLAlchemyのsessionmaker()を呼び出すことで、データベースセッションを管理するためのセッションファクトリオブジェクトを作成します。
__enter__()
def __enter__(self):
self.session = self.session_maker()
return self
セッションを作成したり、ファイルを開いたりする等、コンテキストを開始するときに実行する操作を定義するメソッドです。
__enter__()もWith句（with SessionManager() as manager）のタイミングで実行されますが、前述した__init__()の処理が終わった後に実行されます。
__enter__()内では、__init__()で作成したセッションファクトリを使ってデータベースセッションを開きます。
なお、with句の中でas文を使うと、__enter__()の戻り値を変数として保持することができます。
with SessionManager() as manager:
# manager = SessionManagerインスタンス
上記の例では、SessionManagerインスタンスをmanager変数として保持することで、呼び出し元でsession()やcommit()等のメソッドを利用することができます。
__exit__()
def __exit__(self, exc_type, exc_value, traceback):
if exc_type is not None:
self.session.rollback()
self.session.close()
セッションやファイルを閉じる等、コンテキストを終了するときに実行する操作を定義するメソッドです。
呼び出し元のWith句を抜けたタイミングで実行されます。
このメソッドは以下の３つの引数を利用することができ、コンテキストの実行中に発生した例外情報を取得することができます。
※例外が発生しない場合、３つの引数の値はNoneになります。
-
exc_type- 発生した例外の型を取得
-
exc_value- 例外メッセージの値
-
traceback- 例外が発生した場所等のトレースバック情報
本記事の__exit__()では、exc_typeの値を検査し、例外が発生した場合はrollback()で変更内容を戻します。
また、最後にclose()を定義することで、例外の発生有無に関わらずデータベースセッションを閉じます。
commit()とrollback()
def commit(self):
""" SQLAlchemyのcommitメソッドのWrapper """
# コミット前になにかする場合はここに処理を記述する
self.session.commit()
def rollback(self):
""" SQLAlchemyのrollbackメソッドのWrapper """
# ロールバック前になにかする場合はここに処理を記述する
self.session.rollback()
SQLAlchemyのcommit()とrollback()のWrapperメソッドです。
これらは実装必須ではないですが、Wrapperメソッド内にエラーハンドリングやデバッグ出力処理を追加したり、特定の条件でのみcommitを実行する等の追加処理を実装するなど、Wrapperメソッドを実装することで拡張性をもたせることができます。
参考