家計簿アプリを作成する 補足・注意事項①【Sessionと排他制御】

Python
2025.05.31

SQLAlchemyのEngine作成時の補足

Session

Sessionをインスタンス時にself.engineのみ渡すと、Session.close()した後にデータベースから取得した情報を参照すると「Instance is not bound to a Session; attribute refresh operation cannot proceed（インスタンスはセッションにバインドされていません。属性の更新操作を続行できません。）」というエラーが発生する

これは、参照した時にSQLAlchemyが自動的にデータベースに対してSELECTクエリを発行し、最新のデータを取得してオブジェクトの状態を更新しようとするがSession.close()していて取得できないために発生する

Session.close()後に取得した情報を参照して利用するには、expire_on_commit=Falseを引数に追加する（デフォルトはexpire_on_commit=True）

from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import Session
from db.models import Log
from common.message import Message
from common.create_exception_log import Create_Exception_Log
from common.const import Const
from common.message_box import Message_Box


# DB接続
class Engine:
    def __init__(self):
        self.engine = create_engine("sqlite:///db/example.db", echo=True)
        self.inspector = inspect(self.engine)
        # セッションクラス
        # 2025/05/30 expire_on_commit=False 追加
        self.session = Session(self.engine, expire_on_commit=False)
        # ログクラス
        self.log_row = Log()
        # メッセージクラス
        self.message = Message()
        # 例外エラー出力クラス
        self.create_exception_log = Create_Exception_Log()
        # 定数クラス
        self.const = Const()

アダプター作成時の注意事項

排他制御

UPDATE処理を行う時の楽観的排他制御で以下のソースで制御されなかった
Aユーザーが画面表示→Bユーザーが画面表示し更新→Aユーザーが更新した場合、Bユーザーの更新が成功し、更新時間が変更される
Aユーザーの更新は、1行目を取得するときに同じ更新時間でないためにNoneが返ってきて、IF文で排他エラーメッセージが返される

# 取得条件に
# ユーザーIDが引数で受け取ったユーザーID
# 更新時間が引数で受け取った更新時間(画面に表示時にSELECTで取得した更新時間)
# を追加
stmt = select(User).where(
    User.user_id == arg_user_row.user_id,
    User.update_seq == arg_user_row.update_seq,
)
try:
    # 1行目を取得する
    fill_user = self.session.scalars(stmt).first()
    # 取得した情報がある場合
    if fill_user is not None:
        fill_user.del_flg = "1"
        self.session.commit()
    else:
        #排他エラーメッセージを返す

以上の状態を解消するため、更新バージョンを管理する列を追加
更新バージョンが一致する場合に1行目が取得され、更新できるように対応する

# 取得条件に
# ユーザーIDが引数で受け取ったユーザーID
# 更新時間が引数で受け取った更新時間(画面に表示時にSELECTで取得した更新時間)
# を追加
stmt = select(User).where(
    User.user_id == arg_user_row.user_id,
    User.update_version == arg_user_row.update_version,
)
try:
    # 1行目を取得する
    fill_user = self.session.scalars(stmt).first()
    # 取得した情報がある場合
    if fill_user is not None:
        fill_user.del_flg = "1"
        fill_user.update_version = fill_user.update_version + 1
        self.session.commit()
    else:
        #排他エラーメッセージを返す

全体コード（GitHub）

GitHub - SakumaTakayuki/household_account_book
Contribute to SakumaTakayuki/household_account_book development by creating an account on GitHub.
github.com