SqlAlchemy 2.0 中文文档（二十三）（4）

Session = sessionmaker()
with Session.begin() as session:
    session.add(u1)
    session.add(u2)
    nested = session.begin_nested()  # establish a savepoint
    session.add(u3)
    nested.rollback()  # rolls back u3, keeps u1 and u2
# commits u1 and u2

for record in records:
    try:
        with session.begin_nested():
            session.merge(record)
    except:
        print("Skipped record %s" % record)
session.commit()

from sqlalchemy import exc
with session.begin():
    for record in records:
        try:
            with session.begin_nested():
                obj = SomeRecord(id=record["identifier"], name=record["name"])
                session.add(obj)
        except exc.IntegrityError:
            print(f"Skipped record {record} - row already exists")

ORM                                           Core
-----------------------------------------     -----------------------------------
sessionmaker                                  Engine
Session                                       Connection
sessionmaker.begin()                          Engine.begin()
some_session.commit()                         some_connection.commit()
with some_sessionmaker() as session:          with some_engine.connect() as conn:
with some_sessionmaker.begin() as session:    with some_engine.begin() as conn:
with some_session.begin_nested() as sp:       with some_connection.begin_nested() as sp:

engine = create_engine("postgresql+psycopg2://user:pass@host/dbname")
with engine.connect() as conn:
    conn.execute(
        some_table.insert(),
        [
            {"data": "some data one"},
            {"data": "some data two"},
            {"data": "some data three"},
        ],
    )
    conn.commit()

Session = sessionmaker(engine)
with Session() as session:
    session.add_all(
        [
            SomeClass(data="some data one"),
            SomeClass(data="some data two"),
            SomeClass(data="some data three"),
        ]
    )
    session.commit()

engine = create_engine("postgresql+psycopg2://user:pass@host/dbname")
with engine.begin() as conn:
    conn.execute(
        some_table.insert(),
        [
            {"data": "some data one"},
            {"data": "some data two"},
            {"data": "some data three"},
        ],
    )
# commits and closes automatically

Session = sessionmaker(engine)
with Session.begin() as session:
    session.add_all(
        [
            SomeClass(data="some data one"),
            SomeClass(data="some data two"),
            SomeClass(data="some data three"),
        ]
    )
# commits and closes automatically

engine = create_engine("postgresql+psycopg2://user:pass@host/dbname")
with engine.begin() as conn:
    savepoint = conn.begin_nested()
    conn.execute(
        some_table.insert(),
        [
            {"data": "some data one"},
            {"data": "some data two"},
            {"data": "some data three"},
        ],
    )
    savepoint.commit()  # or rollback
# commits automatically

Session = sessionmaker(engine)
with Session.begin() as session:
    savepoint = session.begin_nested()
    session.add_all(
        [
            SomeClass(data="some data one"),
            SomeClass(data="some data two"),
            SomeClass(data="some data three"),
        ]
    )
    savepoint.commit()  # or rollback
# commits automatically
```### 显式开始
`Session` 具有“自动开始”行为，这意味着一旦操作开始进行，它就会确保存在一个用于跟踪正在进行的操作的 `SessionTransaction`。当调用 `Session.commit()` 时，该事务将被完成。
通常希望特别是在框架集成中，控制“开始”操作发生的时机。为此，`Session` 使用“自动开始”策略，使得可以直接调用 `Session.begin()` 方法来为尚未启动事务的 `Session` 开始事务：
```py
Session = sessionmaker(bind=engine)
session = Session()
session.begin()
try:
    item1 = session.get(Item, 1)
    item2 = session.get(Item, 2)
    item1.foo = "bar"
    item2.bar = "foo"
    session.commit()
except:
    session.rollback()
    raise

Session = sessionmaker(bind=engine)
session = Session()
with session.begin():
    item1 = session.get(Item, 1)
    item2 = session.get(Item, 2)
    item1.foo = "bar"
    item2.bar = "foo"

engine1 = create_engine("postgresql+psycopg2://db1")
engine2 = create_engine("postgresql+psycopg2://db2")
Session = sessionmaker(twophase=True)
# bind User operations to engine 1, Account operations to engine 2
Session.configure(binds={User: engine1, Account: engine2})
session = Session()
# .... work with accounts and users
# commit.  session will issue a flush to all DBs, and a prepare step to all DBs,
# before committing both transactions
session.commit()
```### 设置事务隔离级别 / DBAPI AUTOCOMMIT
大多数 DBAPI 支持可配置的事务隔离级别概念。传统上有四个级别：“READ UNCOMMITTED”、“READ COMMITTED”、“REPEATABLE READ”和“SERIALIZABLE”。这些通常在 DBAPI 连接开始新事务之前应用，需要注意的是，大多数 DBAPI 在首次发出 SQL 语句时会隐式开始此事务。
支持隔离级别的 DBAPI 通常还支持真实的“自动提交”概念，这意味着 DBAPI 连接本身将被放置在非事务自动提交模式中。这通常意味着自动向数据库发出“BEGIN”的典型 DBAPI 行为不再发生，但也可能包括其他指令。在使用此模式时，**DBAPI 在任何情况下都不使用事务**。SQLAlchemy 方法如`.begin()`、`.commit()` 和 `.rollback()` 会静默通过。
SQLAlchemy 的方言支持在每个`Engine` 或每个`Connection` 的基础上设置隔离模式，使用`create_engine()` 层次上以及 `Connection.execution_options()` 层次上的标志。
当使用 ORM `Session` 时，它充当引擎和连接的*外观*，但不直接暴露事务隔离。因此，为了影响事务隔离级别，我们需要在适当时对`Engine` 或 `Connection` 进行操作。
另请参阅
设置事务隔离级别，包括 DBAPI 自动提交 - 一定要查看 SQLAlchemy `Connection` 对象级别的隔离级别是如何工作的。
#### 为 Sessionmaker / Engine 设置隔离级别
要为 `Session` 或 `sessionmaker` 设置特定的隔离级别，全局首选技巧是可以始终根据特定的隔离级别构建一个 `Engine`，然后将其用作 `Session` 和/或 `sessionmaker` 的连接源：
```py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
eng = create_engine(
    "postgresql+psycopg2://scott:tiger@localhost/test",
    isolation_level="REPEATABLE READ",
)
Session = sessionmaker(eng)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
eng = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
autocommit_engine = eng.execution_options(isolation_level="AUTOCOMMIT")
transactional_session = sessionmaker(eng)
autocommit_session = sessionmaker(autocommit_engine)

with autocommit_session() as session:
    some_objects = session.execute(text("<statement>"))
    some_other_objects = session.execute(text("<statement>"))
# closes connection

plain_engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
autocommit_engine = plain_engine.execution_options(isolation_level="AUTOCOMMIT")
# will normally use plain_engine
Session = sessionmaker(plain_engine)
# make a specific Session that will use the "autocommit" engine
with Session(bind=autocommit_engine) as session:
    # work with session
    ...

with Session() as session:
    session.bind_mapper(User, autocommit_engine)

from sqlalchemy.orm import Session
# assume session just constructed
sess = Session(bind=engine)
# call connection() with options before any other operations proceed.
# this will procure a new connection from the bound engine and begin a real
# database transaction.
sess.connection(execution_options={"isolation_level": "SERIALIZABLE"})
# ... work with session in SERIALIZABLE isolation level...
# commit transaction.  the connection is released
# and reverted to its previous isolation level.
sess.commit()
# subsequent to commit() above, a new transaction may be begun if desired,
# which will proceed with the previous default isolation level unless
# it is set again.

sess = Session(bind=engine)
with sess.begin():
    # call connection() with options before any other operations proceed.
    # this will procure a new connection from the bound engine and begin a
    # real database transaction.
    sess.connection(execution_options={"isolation_level": "SERIALIZABLE"})
    # ... work with session in SERIALIZABLE isolation level...
# outside the block, the transaction has been committed.  the connection is
# released and reverted to its previous isolation level.

Session = sessionmaker()
with Session.begin() as session:
    session.add(u1)
    session.add(u2)
    nested = session.begin_nested()  # establish a savepoint
    session.add(u3)
    nested.rollback()  # rolls back u3, keeps u1 and u2
# commits u1 and u2

for record in records:
    try:
        with session.begin_nested():
            session.merge(record)
    except:
        print("Skipped record %s" % record)
session.commit()

from sqlalchemy import exc
with session.begin():
    for record in records:
        try:
            with session.begin_nested():
                obj = SomeRecord(id=record["identifier"], name=record["name"])
                session.add(obj)
        except exc.IntegrityError:
            print(f"Skipped record {record} - row already exists")