SQLAlchemyのトランザクションについて

Question

前提・実現したいこと

バッチを複数並列に実行することを想定しています。
その場合、DBへの書き込みは、session.commit()すればいいだけなのでしょうか？
マルチスレッドでも問題なく動作し、適切な場所でcommitorrollbackされるのでしょうか？
サンプルコードを作成して試してみて、実現できているように後述する疑問が残ってしましました。

試したこと

下記のようにsetting.pyにsessonを定義し
db関連をまとめたパッケージの__init__.pyでimportしています

# setting.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session

# DB設定
USERNAME = "test"
PASSWORD = "test"
HOST = "127.0.0.1"
DBNAME = "TEST"
CHARSET = "utf8"
ENCODING = "utf-8"
DB_ECHO = True

# mysql
DATABASE = f"mysql://{USERNAME}:{PASSWORD}@{HOST}/{DBNAME}?charset={CHARSET}"

ENGINE = create_engine(
    DATABASE,
    encoding=ENCODING,
    echo=DB_ECHO
)

# session作成
session = scoped_session(
    sessionmaker(
        autocommit=False,
        autoflush=False,
        bind=ENGINE
    )
)

DbBase = declarative_base()

マルチスレッドで2テーブルにinsertするテストコードを実行してみました

# main
from db import DbBase, session, ENGINE
from db.job1 import Job1
from db.job2 import Job2

import threading
import time

DbBase.metadata.drop_all(bind=ENGINE, tables=[Job1.__table__, Job2.__table__])
DbBase.metadata.create_all(bind=ENGINE, tables=[Job1.__table__, Job2.__table__])

def create_sample1():

    for i in range(1,1000):
        job = Job1()
        job.name = f"TEST{i}"
        job.note = i
        session.add(job)
        time.sleep(0.01)

    session.rollback()


def create_sample2():

    for i in range(1, 1000):
        job = Job2()
        job.name = f"TEST{i}"
        session.add(job)
        time.sleep(0.02)

    session.commit()


thread1 = threading.Thread(target=create_sample1)
thread2 = threading.Thread(target=create_sample2)


thread1.start()
thread2.start()

試した上での疑問

job1のほうはrollbackされ、job2はcommitされており実現はされているような気はします。
ただ、上記ソースではトランザクションが始まるタイミングがわからず、本当に大丈夫なのか？という不安も解消されていません。

sessionはグローバルで定義されており、共通に利用しています。
それなのに、なぜ片方のみrollbackができるのでしょうか？どこからトランザクションが開始されるのでしょうか？

※ トランザクション開始を明示するならと
session.begin()と書くとautocommit=Falseの時は、トランザクションがすでに貼られていますと言われ
session.begin_nested()と書くと、おそらくmysqlには対応していないのかinsertが反映されてませんでした。

もやもやっとした疑問のため、わかりづらい箇所があるかと思いますが
よろしくお願いします。

Answer 1

解決しました。

参考：
https://stackoverflow.com/questions/6297404/multi-threaded-use-of-sqlalchemy

理解した内容を要約すると
・Sessionオブジェクトはスレッドセーフではない
・マルチスレッドで使いたければ、スレッドローカルストレージを使いなさい
・scoped_sessionを使うと、デフォルトでスレッドローカルストレージを使っています
・そのため、スレッド単位で1つのセッションです
・トランザクションは、commitorrollback直後に新しいのが張られます

上記ソースで、トランザクションは問題ないという結論になりました。