SQLAlchemy에서 Transaction사용하기

SQLAlchemy에서는 Transaction을 사용하여 with 구문이 정상 실행시에 commit되고 구문내에서 예외 발생시에 rollback 되도록 할 수 있다.

Commit As You Go

SQLAlchemy 2.0 이상부터 제공되는 기능. 공식문서에 아래와 같이 안내되어 있다.

1.4 버전에서 테스트 해보니, commit()실행과 상관없이 각 실행이 autocommit되며 connection에서 commit()을 지원하지 않는다.

New in version 2.0: “commit as you go” style is a new feature of SQLAlchemy 2.0. It is also available in SQLAlchemy 1.4’s “transitional” mode when using a “future” style engine.

from sqlalchemy import create_engine
from sqlalchemy import text

engine = create_engine("{db_url}")
with engine.connect() as connection:
	  connection.execute(text("INSERT into test (id) values (1)"))
    connection.execute(text("INSERT into test (id) values (2)"))
		# commit을 명시적으로 실행 해주어야 함
		connection.commit()

Begin Once

sqlalchemy==1.4.x

from sqlalchemy import create_engine

engine = create_engine("{db_url}")
with engine.connect() as connection:
    with connection.begin():
        connection.execute("INSERT into test (id) values (1)")
        connection.execute("INSERT into test (id) values (2)")

sqlalchemy==2.0.x

from sqlalchemy import create_engine
from sqlalchemy import text

engine = create_engine("{db_url}")
with engine.connect() as connection:
    with connection.begin():
        connection.execute(text("INSERT into test (id) values (1)"))
        connection.execute(text("INSERT into test (id) values (2)"))

Connect and Begin Once from the Engine

sqlalchemy==1.4.x

from sqlalchemy import create_engine

engine = create_engine("{db_url}")
with warehouse_engine.begin() as connection:
    connection.execute("INSERT into test (id) values (1)")
    connection.execute("INSERT into test (id) values (2)")

sqlalchemy==2.0.x

from sqlalchemy import create_engine
from sqlalchemy import text

engine = create_engine("{db_url}")
with warehouse_engine.begin() as connection:
    connection.execute(text("INSERT into test (id) values (1)"))
    connection.execute(text("INSERT into test (id) values (2)"))

with문 대신 try/except 문 사용

sqlalchemy==1.4.x

from sqlalchemy import create_engine

engine = create_engine("{db_url}")
connection = warehouse_engine.connect()
trans = connection.begin()
try:
    connection.execute("INSERT into test (id) values (1)")
		connection.execute("INSERT into test (id) values (2)")
    trans.commit()
except:
    trans.rollback()
    print("rollback")
    raise
finally:
    connection.close()

sqlalchemy==2.0.x

from sqlalchemy import create_engine
from sqlalchemy import text

engine = create_engine("{db_url}")
connection = warehouse_engine.connect()
trans = connection.begin()
try:
    connection.execute(text("INSERT into test (id) values (1)"))
    connection.execute(text("INSERT into test (id) values (2)"))
    trans.commit()
except:
    trans.rollback()
    print("rollback")
    raise
finally:
    connection.close()