SQLAlchemy nested transactions

sqlalchemy_nested_example.py

"""
This script demonstrates the use of nested transactions in SQLAlchemy, including
a workaround for an issue with the SQLite backend.

References:
http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#using-savepoint
http://docs.sqlalchemy.org/en/latest/dialects/sqlite.html#serializable-isolation-savepoints-transactional-ddl
"""

from sqlalchemy import Column, String, Integer
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# log SQL statements
import logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

Base = declarative_base()

class Person(Base):
    __tablename__ = 'people'
    id = Column(Integer, primary_key=True)
    name = Column(String)

path = 'test2.db'
engine = create_engine('sqlite:///{}'.format(path))

@event.listens_for(engine, "connect")
def do_connect(dbapi_connection, connection_record):
    # disable pysqlite's emitting of the BEGIN statement entirely.
    # also stops it from emitting COMMIT before any DDL.
    dbapi_connection.isolation_level = None

@event.listens_for(engine, "begin")
def do_begin(conn):
    # emit our own BEGIN
    conn.execute("BEGIN")

# create all tables
Base.metadata.create_all(engine)

Session = sessionmaker()
Session.configure(bind=engine)
session = Session()

u1 = Person(name='Dave')
session.add(u1)

# a nested section
session.begin_nested()

# another nested section
session.begin_nested()
u2 = Person(name='Steve')
session.add(u2)
session.commit()

session.rollback()

session.commit()

session.close()