中文
devexamples

SQLAlchemy Transaction Isolation Level

Set per-session isolation levels in SQLAlchemy, using repeatable read for consistent reports and read committed for regular traffic.

PythonData Integrity

Code

snippet
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

engine = create_engine(
    'postgresql+psycopg://user:pass@localhost/reporting',
    isolation_level='READ COMMITTED',
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
)

reporting_engine = engine.execution_options(
    isolation_level='REPEATABLE READ'
)


def transfer_funds(session: Session, src: int, dst: int, amount: int) -> None:
    session.execute(
        text('UPDATE accounts SET balance = balance - :amt WHERE id = :id'),
        {'amt': amount, 'id': src},
    )
    row = session.execute(
        text('SELECT balance FROM accounts WHERE id = :id'),
        {'id': src},
    ).scalar_one()
    if row < 0:
        raise ValueError('overdraft rejected')
    session.execute(
        text('UPDATE accounts SET balance = balance + :amt WHERE id = :id'),
        {'amt': amount, 'id': dst},
    )


with Session(engine) as session:
    with session.begin():
        transfer_funds(session, src=1, dst=2, amount=75)

with Session(reporting_engine) as session:
    snapshot = session.execute(
        text('SELECT sum(balance) FROM accounts')
    ).scalar_one()
    print('consistent snapshot total:', snapshot)

Related Examples