SqlAlchemy 2.0 中文文档（五十六）

from sqlalchemy import column
from sqlalchemy import create_engine
from sqlalchemy import select
from sqlalchemy import table

engine = create_engine("sqlite://")

engine.execute("CREATE TABLE foo (id integer)")
engine.execute("INSERT INTO foo (id) VALUES (1)")

foo = table("foo", column("id"))
result = engine.execute(select([foo.c.id]))

print(result.fetchall())

$ python test3.py
[(1,)]

SQLALCHEMY_WARN_20=1 python -W always::DeprecationWarning test3.py

$ SQLALCHEMY_WARN_20=1 python2 -W always::DeprecationWarning test3.py
test3.py:9: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0\. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  engine.execute("CREATE TABLE foo (id integer)")
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:2856: RemovedIn20Warning: Passing a string to Connection.execute() is deprecated and will be removed in version 2.0\.  Use the text() construct, or the Connection.exec_driver_sql() method to invoke a driver-level SQL string. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  return connection.execute(statement, *multiparams, **params)
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:1639: RemovedIn20Warning: The current statement is being autocommitted using implicit autocommit.Implicit autocommit will be removed in SQLAlchemy 2.0\.   Use the .begin() method of Engine or Connection in order to use an explicit transaction for DML and DDL statements. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  self._commit_impl(autocommit=True)
test3.py:10: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0\. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  engine.execute("INSERT INTO foo (id) VALUES (1)")
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:2856: RemovedIn20Warning: Passing a string to Connection.execute() is deprecated and will be removed in version 2.0\.  Use the text() construct, or the Connection.exec_driver_sql() method to invoke a driver-level SQL string. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  return connection.execute(statement, *multiparams, **params)
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:1639: RemovedIn20Warning: The current statement is being autocommitted using implicit autocommit.Implicit autocommit will be removed in SQLAlchemy 2.0\.   Use the .begin() method of Engine or Connection in order to use an explicit transaction for DML and DDL statements. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  self._commit_impl(autocommit=True)
/home/classic/dev/sqlalchemy/lib/sqlalchemy/sql/selectable.py:4271: RemovedIn20Warning: The legacy calling style of select() is deprecated and will be removed in SQLAlchemy 2.0\.  Please use the new calling style described at select(). (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  return cls.create_legacy_select(*args, **kw)
test3.py:14: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0\. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  result = engine.execute(select([foo.c.id]))
[(1,)]

from sqlalchemy import column
from sqlalchemy import create_engine
from sqlalchemy import select
from sqlalchemy import table
from sqlalchemy import text

engine = create_engine("sqlite://")

# don't rely on autocommit for DML and DDL
with engine.begin() as connection:
    # use connection.execute(), not engine.execute()
    # use the text() construct to execute textual SQL
    connection.execute(text("CREATE TABLE foo (id integer)"))
    connection.execute(text("INSERT INTO foo (id) VALUES (1)"))

foo = table("foo", column("id"))

with engine.connect() as connection:
    # use connection.execute(), not engine.execute()
    # select() now accepts column / table expressions positionally
    result = connection.execute(select(foo.c.id))

    print(result.fetchall())

import warnings
from sqlalchemy import exc

# for warnings not included in regex-based filter below, just log
warnings.filterwarnings("always", category=exc.RemovedIn20Warning)

# for warnings related to execute() / scalar(), raise
for msg in [
    r"The (?:Executable|Engine)\.(?:execute|scalar)\(\) function",
    r"The current statement is being autocommitted using implicit autocommit,",
    r"The connection.execute\(\) method in SQLAlchemy 2.0 will accept "
    "parameters as a single dictionary or a single sequence of "
    "dictionaries only.",
    r"The Connection.connect\(\) function/method is considered legacy",
    r".*DefaultGenerator.execute\(\)",
]:
    warnings.filterwarnings(
        "error",
        message=msg,
        category=exc.RemovedIn20Warning,
    )

from sqlalchemy import create_engine

engine = create_engine("postgresql+psycopg2:///")

with engine.connect() as conn:
    conn.execute(text("insert into table (x) values (:some_x)"), {"some_x": 10})

    conn.commit()  # commit as you go

from sqlalchemy.orm import Session

with Session(engine) as session:
    session.add(MyObject())
    session.commit()

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)

with Session.begin() as session:
    session.add(MyObject())

Base = declarative_base()

class Foo(Base):
    __tablename__ = "foo"

    id: int = Column(Integer, primary_key=True)

    # will raise
    bars: List["Bar"] = relationship("Bar", back_populates="foo")

class Bar(Base):
    __tablename__ = "bar"

    id: int = Column(Integer, primary_key=True)
    foo_id = Column(ForeignKey("foo.id"))

    # will raise
    foo: Foo = relationship(Foo, back_populates="bars", cascade="all")

# qualify the base with __allow_unmapped__.  Can also be
# applied to classes directly if preferred
class Base:
    __allow_unmapped__ = True

Base = declarative_base(cls=Base)

# existing mapping proceeds, Declarative will ignore any annotations
# which don't include ``Mapped[]``
class Foo(Base):
    __tablename__ = "foo"

    id: int = Column(Integer, primary_key=True)

    bars: List["Bar"] = relationship("Bar", back_populates="foo")

class Bar(Base):
    __tablename__ = "bar"

    id: int = Column(Integer, primary_key=True)
    foo_id = Column(ForeignKey("foo.id"))

    foo: Foo = relationship(Foo, back_populates="bars", cascade="all")

conn = engine.connect()

# won't autocommit in 2.0
conn.execute(some_table.insert().values(foo="bar"))

conn = engine.connect()

# won't autocommit in 2.0
conn.execute(text("INSERT INTO table (foo) VALUES ('bar')"))

conn = engine.connect()

# won't autocommit in 2.0
conn.execute(text("EXEC my_procedural_thing()").execution_options(autocommit=True))

with engine.begin() as conn:
    conn.execute(some_table.insert().values(foo="bar"))
    conn.execute(some_other_table.insert().values(bat="hoho"))

with engine.connect() as conn:
    with conn.begin():
        conn.execute(some_table.insert().values(foo="bar"))
        conn.execute(some_other_table.insert().values(bat="hoho"))

with engine.begin() as conn:
    conn.execute(text("EXEC my_procedural_thing()"))

with engine.connect() as conn:
    conn.execute(some_table.insert().values(foo="bar"))
    conn.execute(some_other_table.insert().values(bat="hoho"))

    conn.commit()

with engine.begin() as conn:
    conn.execute(some_table.insert().values(foo="bar"))

# 1.4 / 2.0 code

from sqlalchemy import create_engine

engine = create_engine(..., future=True)

with engine.connect() as conn:
    conn.execute(some_table.insert().values(foo="bar"))
    conn.commit()

    conn.execute(text("some other SQL"))
    conn.rollback()

from sqlalchemy import MetaData

metadata_obj = MetaData(bind=engine)  # no longer supported

metadata_obj.create_all()  # requires Engine or Connection

metadata_obj.reflect()  # requires Engine or Connection

t = Table("t", metadata_obj, autoload=True)  # use autoload_with=engine

result = engine.execute(t.select())  # no longer supported

result = t.select().execute()  # no longer supported

from sqlalchemy import MetaData

metadata_obj = MetaData()

# engine level:

# create tables
metadata_obj.create_all(engine)

# reflect all tables
metadata_obj.reflect(engine)

# reflect individual table
t = Table("t", metadata_obj, autoload_with=engine)

# connection level:

with engine.connect() as connection:
    # create tables, requires explicit begin and/or commit:
    with connection.begin():
        metadata_obj.create_all(connection)

    # reflect all tables
    metadata_obj.reflect(connection)

    # reflect individual table
    t = Table("t", metadata_obj, autoload_with=connection)

    # execute SQL statements
    result = connection.execute(t.select())

result = engine.execute(some_statement)

result = some_statement.execute()

# many choices

# bound metadata?
metadata_obj = MetaData(engine)

# or not?
metadata_obj = MetaData()

# execute from engine?
result = engine.execute(stmt)

# or execute the statement itself (but only if you did
# "bound metadata" above, which means you can't get rid of "bound" if any
# part of your program uses this form)
result = stmt.execute()

# execute from connection, but it autocommits?
conn = engine.connect()
conn.execute(stmt)

# execute from connection, but autocommit isn't working, so use the special
# option?
conn.execution_options(autocommit=True).execute(stmt)

# or on the statement ?!
conn.execute(stmt.execution_options(autocommit=True))

# or execute from connection, and we use explicit transaction?
with conn.begin():
    conn.execute(stmt)

# one choice - work with explicit connection, explicit transaction
# (there remain a few variants on how to demarcate the transaction)

# "begin once" - one transaction only per checkout
with engine.begin() as conn:
    result = conn.execute(stmt)

# "commit as you go" - zero or more commits per checkout
with engine.connect() as conn:
    result = conn.execute(stmt)
    conn.commit()

# "commit as you go" but with a transaction block instead of autobegin
with engine.connect() as conn:
    with conn.begin():
        result = conn.execute(stmt)

connection = engine.connect()

# direct string SQL not supported; use text() or exec_driver_sql() method
result = connection.execute("select * from table")

# positional parameters no longer supported, only named
# unless using exec_driver_sql()
result = connection.execute(table.insert(), ("x", "y", "z"))

# **kwargs no longer accepted, pass a single dictionary
result = connection.execute(table.insert(), x=10, y=5)

# multiple *args no longer accepted, pass a list
result = connection.execute(
    table.insert(), {"x": 10, "y": 5}, {"x": 15, "y": 12}, {"x": 9, "y": 8}
)

connection = engine.connect()

from sqlalchemy import text

result = connection.execute(text("select * from table"))

# pass a single dictionary for single statement execution
result = connection.execute(table.insert(), {"x": 10, "y": 5})

# pass a list of dictionaries for executemany
result = connection.execute(
    table.insert(), [{"x": 10, "y": 5}, {"x": 15, "y": 12}, {"x": 9, "y": 8}]
)

engine = create_engine(..., future=True)  # using future mode

with engine.connect() as conn:
    result = conn.execute(text("select x, y from table"))

    row = result.first()  # suppose the row is (1, 2)

    "x" in row  # evaluates to False, in 1.x / future=False, this would be True

    1 in row  # evaluates to True, in 1.x / future=False, this would be False

from sqlalchemy.future.orm import Session

session = Session(some_engine)

result = session.execute(stmt)
for row in result.mappings():
    print("the user is: %s" % row["User"])

from sqlalchemy.future import select

stmt = select(User, Address).join(User.addresses)

for row in session.execute(stmt).mappings():
    print("the user is: %s the address is: %s" % (row[User], row[Address]))

# select_from / order_by keywords no longer supported
stmt = select([1], select_from=table, order_by=table.c.id)

# whereclause parameter no longer supported
stmt = select([table.c.x], table.c.id == 5)

# whereclause parameter no longer supported
stmt = table.select(table.c.id == 5)

# list emits a deprecation warning
stmt = select([table.c.x, table.c.y])

# list emits a deprecation warning
case_clause = case(
    [(table.c.x == 5, "five"), (table.c.x == 7, "seven")],
    else_="neither five nor seven",
)

# use generative methods
stmt = select(1).select_from(table).order_by(table.c.id)

# use generative methods
stmt = select(table).where(table.c.id == 5)

# use generative methods
stmt = table.select().where(table.c.id == 5)

# pass columns clause expressions positionally
stmt = select(table.c.x, table.c.y)

# case conditions passed positionally
case_clause = case(
    (table.c.x == 5, "five"), (table.c.x == 7, "seven"), else_="neither five nor seven"
)

# table columns for CREATE TABLE - structural
table = Table("table", metadata_obj, Column("x", Integer), Column("y", Integer))

# columns in a SELECT statement - structural
stmt = select(table.c.x, table.c.y)

# literal elements in an IN clause - data
stmt = stmt.where(table.c.y.in_([1, 2, 3]))

# no longer supported
stmt = insert(table, values={"x": 10, "y": 15}, inline=True)

# no longer supported
stmt = insert(table, values={"x": 10, "y": 15}, returning=[table.c.x])

# no longer supported
stmt = table.delete(table.c.x > 15)

# no longer supported
stmt = table.update(table.c.x < 15, preserve_parameter_order=True).values(
    [(table.c.y, 20), (table.c.x, table.c.y + 10)]
)

# use generative methods, **kwargs OK for values()
stmt = insert(table).values(x=10, y=15).inline()

# use generative methods, dictionary also still  OK for values()
stmt = insert(table).values({"x": 10, "y": 15}).returning(table.c.x)

# use generative methods
stmt = table.delete().where(table.c.x > 15)

# use generative methods, ordered_values() replaces preserve_parameter_order
stmt = (
    table.update()
    .where(
        table.c.x < 15,
    )
    .ordered_values((table.c.y, 20), (table.c.x, table.c.y + 10))
)

from sqlalchemy.ext import declarative_base, declared_attr

from sqlalchemy.orm import declarative_base, declared_attr

from sqlalchemy.orm import mapper

mapper(SomeClass, some_table, properties={"related": relationship(SomeRelatedClass)})

from sqlalchemy.orm import registry

mapper_reg = registry()

mapper_reg.map_imperatively(
    SomeClass, some_table, properties={"related": relationship(SomeRelatedClass)}
)

from sqlalchemy.orm import registry

mapper_reg = registry()

Base = mapper_reg.generate_base()

class SomeRelatedClass(Base):
    __tablename__ = "related"

    # ...

mapper_reg.map_imperatively(
    SomeClass,
    some_table,
    properties={
        "related": relationship(
            "SomeRelatedClass",
            primaryjoin="SomeRelatedClass.related_id == SomeClass.id",
        )
    },
)

session.query(User).get(42)

session.get(User, 42)

session.query(User).all()

session.execute(
  select(User)
).scalars().all()

# or

session.scalars(
  select(User)
).all()

session.query(User).\
  filter_by(name="some user").\
  one()

session.execute(
  select(User).
  filter_by(name="some user")
).scalar_one()

session.query(User).\
  filter_by(name="some user").\
  first()

session.scalars(
  select(User).
  filter_by(name="some user").
  limit(1)
).first()

session.query(User).options(
  joinedload(User.addresses)
).all()

session.scalars(
  select(User).
  options(
    joinedload(User.addresses)
  )
).unique().all()

session.query(User).\
  join(Address).\
  filter(
    Address.email == "e@sa.us"
  ).\
  all()

session.execute(
  select(User).
  join(Address).
  where(
    Address.email == "e@sa.us"
  )
).scalars().all()

session.query(User).\
  from_statement(
    text("select * from users")
  ).\
  all()

session.scalars(
  select(User).
  from_statement(
    text("select * from users")
  )
).all()

session.query(User).\
  join(User.addresses).\
  options(
    contains_eager(User.addresses)
  ).\
  populate_existing().all()

session.execute(
  select(User)
  .join(User.addresses)
  .options(
    contains_eager(User.addresses)
  )
  .execution_options(
      populate_existing=True
  )
).scalars().all()

session.query(User).\
  filter(User.name == "foo").\
  update(
    {"fullname": "Foo Bar"},
    synchronize_session="evaluate"
  )

session.execute(
  update(User)
  .where(User.name == "foo")
  .values(fullname="Foo Bar")
  .execution_options(
    synchronize_session="evaluate"
  )
)

session.query(User).count()

session.scalar(
  select(func.count()).
  select_from(User)
)

# or

session.scalar(
  select(func.count(User.id))
)

session = Session(engine)

# becomes legacy use case
user = session.query(User).filter_by(name="some user").one()

# becomes legacy use case
user = session.query(User).filter_by(name="some user").first()

# becomes legacy use case
user = session.query(User).get(5)

# becomes legacy use case
for user in (
    session.query(User).join(User.addresses).filter(Address.email == "some@email.com")
):
    ...

# becomes legacy use case
users = session.query(User).options(joinedload(User.addresses)).order_by(User.id).all()

# becomes legacy use case
users = session.query(User).from_statement(text("select * from users")).all()

# etc

session = Session(engine)

user = session.execute(select(User).filter_by(name="some user")).scalar_one()

# for first(), no LIMIT is applied automatically; add limit(1) if LIMIT
# is desired on the query
user = (
    session.execute(select(User).filter_by(name="some user").limit(1)).scalars().first()
)

# get() moves to the Session directly
user = session.get(User, 5)

for user in session.execute(
    select(User).join(User.addresses).filter(Address.email == "some@email.case")
).scalars():
    ...

# when using joinedload() against collections, use unique() on the result
users = (
    session.execute(select(User).options(joinedload(User.addresses)).order_by(User.id))
    .unique()
    .all()
)

# select() has ORM-ish methods like from_statement() that only work
# if the statement is against ORM entities
users = (
    session.execute(select(User).from_statement(text("select * from users")))
    .scalars()
    .all()
)

# legacy usage
user_obj = session.query(User).get(5)

# 1.4 / 2.0 cross-compatible use
user_obj = session.get(User, 5)

# string use removed
q = session.query(User).join("addresses")

# string use removed
q = session.query(User).options(joinedload("addresses"))

# string use removed
q = session.query(Address).filter(with_parent(u1, "addresses"))

# compatible with all modern SQLAlchemy versions

q = session.query(User).join(User.addresses)

q = session.query(User).options(joinedload(User.addresses))

q = session.query(Address).filter(with_parent(u1, User.addresses))

# SQLAlchemy 1.4 / 2.0 cross compatible use

stmt = select(User).join(User.addresses)
result = session.execute(stmt)

stmt = select(User).options(joinedload(User.addresses))
result = session.execute(stmt)

stmt = select(Address).where(with_parent(u1, User.addresses))
result = session.execute(stmt)

# chaining removed
q = session.query(User).join("orders", "items", "keywords")

q = session.query(User).join(User.orders).join(Order.items).join(Item.keywords)

# 1.4 / 2.0 cross compatible

stmt = select(User).join(User.orders).join(Order.items).join(Item.keywords)
result = session.execute(stmt)

# join_from can also be helpful
stmt = select(User).join_from(User, Order).join_from(Order, Item, Order.items)
result = session.execute(stmt)

# no longer supported
q = (
    session.query(Node)
    .join("children", aliased=True)
    .filter(Node.name == "some sub child")
    .join("children", from_joinpoint=True, aliased=True)
    .filter(Node.name == "some sub sub child")
)

n1 = aliased(Node)
n2 = aliased(Node)

q = (
    select(Node)
    .join(Node.children.of_type(n1))
    .where(n1.name == "some sub child")
    .join(n1.children.of_type(n2))
    .where(n2.name == "some sub child")
)

# 1.xx code

result = (
    session.query(User)
    .join(User.addresses)
    .distinct()
    .order_by(Address.email_address)
    .all()
)

# 1.4 / 2.0 code

stmt = (
    select(User, Address.email_address)
    .join(User.addresses)
    .distinct()
    .order_by(Address.email_address)
)

result = session.execute(stmt).columns(User).all()

# from_self is removed
q = (
    session.query(User, Address.email_address)
    .join(User.addresses)
    .from_self(User)
    .order_by(Address.email_address)
)

from sqlalchemy.orm import aliased

subq = session.query(User, Address.email_address).join(User.addresses).subquery()

ua = aliased(User, subq)

aa = aliased(Address, subq)

q = session.query(ua, aa).order_by(aa.email_address)

from sqlalchemy.orm import aliased

subq = select(User, Address.email_address).join(User.addresses).subquery()

ua = aliased(User, subq)

aa = aliased(Address, subq)

stmt = select(ua, aa).order_by(aa.email_address)

result = session.execute(stmt)

# 1.4 / 2.0 code

subq = select(User, Address).join(User.addresses).subquery()

ua = aliased(User, subq)
aa = aliased(Address, subq)

stmt = select(ua, aa).order_by(aa.email_address)
result = session.execute(stmt)

SELECT  anon_1.id  AS  anon_1_id,  anon_1.id_1  AS  anon_1_id_1,
  anon_1.user_id  AS  anon_1_user_id,
  anon_1.email_address  AS  anon_1_email_address
FROM  (
  SELECT  "user".id  AS  id,  address.id  AS  id_1,
  address.user_id  AS  user_id,  address.email_address  AS  email_address
  FROM  "user"  JOIN  address  ON  "user".id  =  address.user_id
)  AS  anon_1  ORDER  BY  anon_1.email_address

subquery = session.query(User).filter(User.id == 5).subquery()

user = session.query(User).select_entity_from(subquery).first()

from sqlalchemy.orm import aliased

subquery = session.query(User).filter(User.name.like("%somename%")).subquery()

ua = aliased(User, subquery)

user = session.query(ua).order_by(ua.id).first()

from sqlalchemy.orm import aliased

subquery = select(User).where(User.name.like("%somename%")).subquery()

ua = aliased(User, subquery)

# note that LIMIT 1 is not automatically supplied, if needed
user = session.execute(select(ua).order_by(ua.id).limit(1)).scalars().first()

# In the legacy API, many rows each have the same User primary key, but
# only one User per primary key is returned
users = session.query(User).options(joinedload(User.addresses))

# In the new API, uniquing is available but not implicitly
# enabled
result = session.execute(select(User).options(joinedload(User.addresses)))

# this actually will raise an error to let the user know that
# uniquing should be applied
rows = result.all()

# 1.4 / 2.0 code

stmt = select(User).options(joinedload(User.addresses))

# statement will raise if unique() is not used, due to joinedload()
# of a collection.  in all other cases, unique() is not needed.
# By stating unique() explicitly, confusion over discrepancies between
# number of objects/ rows returned vs. "SELECT COUNT(*)" is resolved
rows = session.execute(stmt).unique().all()

class User(Base):
    __tablename__ = "user"

    posts = relationship(Post, lazy="dynamic")

jack = session.get(User, 5)

# filter Jack's blog posts
posts = session.scalars(jack.posts.statement.where(Post.headline == "this is a post"))

from sqlalchemy.orm import with_parent

jack = session.get(User, 5)

posts = session.scalars(
    select(Post)
    .where(with_parent(jack, User.posts))
    .where(Post.headline == "this is a post")
)

from sqlalchemy.orm import Session

sess = Session(engine, autocommit=True)

# no transaction begun, but emits SQL, won't be supported
obj = sess.query(Class).first()

# session flushes in a transaction that it begins and
# commits, won't be supported
sess.flush()

from sqlalchemy.orm import Session

sess = Session(engine)

sess.begin()  # begin explicitly; if not called, will autobegin
# when database access is needed

sess.add(obj)

sess.commit()

import contextlib

@contextlib.contextmanager
def transaction(session):
    if not session.in_transaction():
        with session.begin():
            yield
    else:
        yield

# method_a starts a transaction and calls method_b
def method_a(session):
    with transaction(session):
        method_b(session)

# method_b also starts a transaction, but when
# called from method_a participates in the ongoing
# transaction.
def method_b(session):
    with transaction(session):
        session.add(SomeObject("bat", "lala"))

Session = sessionmaker(engine)

# create a Session and call method_a
with Session() as session:
    method_a(session)

def method_a(session):
    method_b(session)

def method_b(session):
    session.add(SomeObject("bat", "lala"))

Session = sessionmaker(engine)

# create a Session and call method_a
with Session() as session:
    with session.begin():
        method_a(session)

from sqlalchemy import column
from sqlalchemy import create_engine
from sqlalchemy import select
from sqlalchemy import table

engine = create_engine("sqlite://")

engine.execute("CREATE TABLE foo (id integer)")
engine.execute("INSERT INTO foo (id) VALUES (1)")

foo = table("foo", column("id"))
result = engine.execute(select([foo.c.id]))

print(result.fetchall())

$ python test3.py
[(1,)]

SQLALCHEMY_WARN_20=1 python -W always::DeprecationWarning test3.py

$ SQLALCHEMY_WARN_20=1 python2 -W always::DeprecationWarning test3.py
test3.py:9: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0\. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  engine.execute("CREATE TABLE foo (id integer)")
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:2856: RemovedIn20Warning: Passing a string to Connection.execute() is deprecated and will be removed in version 2.0\.  Use the text() construct, or the Connection.exec_driver_sql() method to invoke a driver-level SQL string. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  return connection.execute(statement, *multiparams, **params)
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:1639: RemovedIn20Warning: The current statement is being autocommitted using implicit autocommit.Implicit autocommit will be removed in SQLAlchemy 2.0\.   Use the .begin() method of Engine or Connection in order to use an explicit transaction for DML and DDL statements. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  self._commit_impl(autocommit=True)
test3.py:10: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0\. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  engine.execute("INSERT INTO foo (id) VALUES (1)")
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:2856: RemovedIn20Warning: Passing a string to Connection.execute() is deprecated and will be removed in version 2.0\.  Use the text() construct, or the Connection.exec_driver_sql() method to invoke a driver-level SQL string. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  return connection.execute(statement, *multiparams, **params)
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:1639: RemovedIn20Warning: The current statement is being autocommitted using implicit autocommit.Implicit autocommit will be removed in SQLAlchemy 2.0\.   Use the .begin() method of Engine or Connection in order to use an explicit transaction for DML and DDL statements. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  self._commit_impl(autocommit=True)
/home/classic/dev/sqlalchemy/lib/sqlalchemy/sql/selectable.py:4271: RemovedIn20Warning: The legacy calling style of select() is deprecated and will be removed in SQLAlchemy 2.0\.  Please use the new calling style described at select(). (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  return cls.create_legacy_select(*args, **kw)
test3.py:14: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0\. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  result = engine.execute(select([foo.c.id]))
[(1,)]

from sqlalchemy import column
from sqlalchemy import create_engine
from sqlalchemy import select
from sqlalchemy import table
from sqlalchemy import text

engine = create_engine("sqlite://")

# don't rely on autocommit for DML and DDL
with engine.begin() as connection:
    # use connection.execute(), not engine.execute()
    # use the text() construct to execute textual SQL
    connection.execute(text("CREATE TABLE foo (id integer)"))
    connection.execute(text("INSERT INTO foo (id) VALUES (1)"))

foo = table("foo", column("id"))

with engine.connect() as connection:
    # use connection.execute(), not engine.execute()
    # select() now accepts column / table expressions positionally
    result = connection.execute(select(foo.c.id))

    print(result.fetchall())

import warnings
from sqlalchemy import exc

# for warnings not included in regex-based filter below, just log
warnings.filterwarnings("always", category=exc.RemovedIn20Warning)

# for warnings related to execute() / scalar(), raise
for msg in [
    r"The (?:Executable|Engine)\.(?:execute|scalar)\(\) function",
    r"The current statement is being autocommitted using implicit autocommit,",
    r"The connection.execute\(\) method in SQLAlchemy 2.0 will accept "
    "parameters as a single dictionary or a single sequence of "
    "dictionaries only.",
    r"The Connection.connect\(\) function/method is considered legacy",
    r".*DefaultGenerator.execute\(\)",
]:
    warnings.filterwarnings(
        "error",
        message=msg,
        category=exc.RemovedIn20Warning,
    )

from sqlalchemy import create_engine

engine = create_engine("postgresql+psycopg2:///")

with engine.connect() as conn:
    conn.execute(text("insert into table (x) values (:some_x)"), {"some_x": 10})

    conn.commit()  # commit as you go

from sqlalchemy.orm import Session

with Session(engine) as session:
    session.add(MyObject())
    session.commit()

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)

with Session.begin() as session:
    session.add(MyObject())

Base = declarative_base()

class Foo(Base):
    __tablename__ = "foo"

    id: int = Column(Integer, primary_key=True)

    # will raise
    bars: List["Bar"] = relationship("Bar", back_populates="foo")

class Bar(Base):
    __tablename__ = "bar"

    id: int = Column(Integer, primary_key=True)
    foo_id = Column(ForeignKey("foo.id"))

    # will raise
    foo: Foo = relationship(Foo, back_populates="bars", cascade="all")

# qualify the base with __allow_unmapped__.  Can also be
# applied to classes directly if preferred
class Base:
    __allow_unmapped__ = True

Base = declarative_base(cls=Base)

# existing mapping proceeds, Declarative will ignore any annotations
# which don't include ``Mapped[]``
class Foo(Base):
    __tablename__ = "foo"

    id: int = Column(Integer, primary_key=True)

    bars: List["Bar"] = relationship("Bar", back_populates="foo")

class Bar(Base):
    __tablename__ = "bar"

    id: int = Column(Integer, primary_key=True)
    foo_id = Column(ForeignKey("foo.id"))

    foo: Foo = relationship(Foo, back_populates="bars", cascade="all")

from sqlalchemy import column
from sqlalchemy import create_engine
from sqlalchemy import select
from sqlalchemy import table

engine = create_engine("sqlite://")

engine.execute("CREATE TABLE foo (id integer)")
engine.execute("INSERT INTO foo (id) VALUES (1)")

foo = table("foo", column("id"))
result = engine.execute(select([foo.c.id]))

print(result.fetchall())

$ python test3.py
[(1,)]

SQLALCHEMY_WARN_20=1 python -W always::DeprecationWarning test3.py

$ SQLALCHEMY_WARN_20=1 python2 -W always::DeprecationWarning test3.py
test3.py:9: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0\. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  engine.execute("CREATE TABLE foo (id integer)")
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:2856: RemovedIn20Warning: Passing a string to Connection.execute() is deprecated and will be removed in version 2.0\.  Use the text() construct, or the Connection.exec_driver_sql() method to invoke a driver-level SQL string. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  return connection.execute(statement, *multiparams, **params)
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:1639: RemovedIn20Warning: The current statement is being autocommitted using implicit autocommit.Implicit autocommit will be removed in SQLAlchemy 2.0\.   Use the .begin() method of Engine or Connection in order to use an explicit transaction for DML and DDL statements. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  self._commit_impl(autocommit=True)
test3.py:10: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0\. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  engine.execute("INSERT INTO foo (id) VALUES (1)")
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:2856: RemovedIn20Warning: Passing a string to Connection.execute() is deprecated and will be removed in version 2.0\.  Use the text() construct, or the Connection.exec_driver_sql() method to invoke a driver-level SQL string. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  return connection.execute(statement, *multiparams, **params)
/home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:1639: RemovedIn20Warning: The current statement is being autocommitted using implicit autocommit.Implicit autocommit will be removed in SQLAlchemy 2.0\.   Use the .begin() method of Engine or Connection in order to use an explicit transaction for DML and DDL statements. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  self._commit_impl(autocommit=True)
/home/classic/dev/sqlalchemy/lib/sqlalchemy/sql/selectable.py:4271: RemovedIn20Warning: The legacy calling style of select() is deprecated and will be removed in SQLAlchemy 2.0\.  Please use the new calling style described at select(). (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  return cls.create_legacy_select(*args, **kw)
test3.py:14: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0\. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  result = engine.execute(select([foo.c.id]))
[(1,)]

from sqlalchemy import column
from sqlalchemy import create_engine
from sqlalchemy import select
from sqlalchemy import table
from sqlalchemy import text

engine = create_engine("sqlite://")

# don't rely on autocommit for DML and DDL
with engine.begin() as connection:
    # use connection.execute(), not engine.execute()
    # use the text() construct to execute textual SQL
    connection.execute(text("CREATE TABLE foo (id integer)"))
    connection.execute(text("INSERT INTO foo (id) VALUES (1)"))

foo = table("foo", column("id"))

with engine.connect() as connection:
    # use connection.execute(), not engine.execute()
    # select() now accepts column / table expressions positionally
    result = connection.execute(select(foo.c.id))

    print(result.fetchall())

import warnings
from sqlalchemy import exc

# for warnings not included in regex-based filter below, just log
warnings.filterwarnings("always", category=exc.RemovedIn20Warning)

# for warnings related to execute() / scalar(), raise
for msg in [
    r"The (?:Executable|Engine)\.(?:execute|scalar)\(\) function",
    r"The current statement is being autocommitted using implicit autocommit,",
    r"The connection.execute\(\) method in SQLAlchemy 2.0 will accept "
    "parameters as a single dictionary or a single sequence of "
    "dictionaries only.",
    r"The Connection.connect\(\) function/method is considered legacy",
    r".*DefaultGenerator.execute\(\)",
]:
    warnings.filterwarnings(
        "error",
        message=msg,
        category=exc.RemovedIn20Warning,
    )

from sqlalchemy import create_engine

engine = create_engine("postgresql+psycopg2:///")

with engine.connect() as conn:
    conn.execute(text("insert into table (x) values (:some_x)"), {"some_x": 10})

    conn.commit()  # commit as you go

from sqlalchemy.orm import Session

with Session(engine) as session:
    session.add(MyObject())
    session.commit()

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)

with Session.begin() as session:
    session.add(MyObject())

Base = declarative_base()

class Foo(Base):
    __tablename__ = "foo"

    id: int = Column(Integer, primary_key=True)

    # will raise
    bars: List["Bar"] = relationship("Bar", back_populates="foo")

class Bar(Base):
    __tablename__ = "bar"

    id: int = Column(Integer, primary_key=True)
    foo_id = Column(ForeignKey("foo.id"))

    # will raise
    foo: Foo = relationship(Foo, back_populates="bars", cascade="all")

# qualify the base with __allow_unmapped__.  Can also be
# applied to classes directly if preferred
class Base:
    __allow_unmapped__ = True

Base = declarative_base(cls=Base)

# existing mapping proceeds, Declarative will ignore any annotations
# which don't include ``Mapped[]``
class Foo(Base):
    __tablename__ = "foo"

    id: int = Column(Integer, primary_key=True)

    bars: List["Bar"] = relationship("Bar", back_populates="foo")

class Bar(Base):
    __tablename__ = "bar"

    id: int = Column(Integer, primary_key=True)
    foo_id = Column(ForeignKey("foo.id"))

    foo: Foo = relationship(Foo, back_populates="bars", cascade="all")

conn = engine.connect()

# won't autocommit in 2.0
conn.execute(some_table.insert().values(foo="bar"))

conn = engine.connect()

# won't autocommit in 2.0
conn.execute(text("INSERT INTO table (foo) VALUES ('bar')"))

conn = engine.connect()

# won't autocommit in 2.0
conn.execute(text("EXEC my_procedural_thing()").execution_options(autocommit=True))

with engine.begin() as conn:
    conn.execute(some_table.insert().values(foo="bar"))
    conn.execute(some_other_table.insert().values(bat="hoho"))

with engine.connect() as conn:
    with conn.begin():
        conn.execute(some_table.insert().values(foo="bar"))
        conn.execute(some_other_table.insert().values(bat="hoho"))

with engine.begin() as conn:
    conn.execute(text("EXEC my_procedural_thing()"))

with engine.connect() as conn:
    conn.execute(some_table.insert().values(foo="bar"))
    conn.execute(some_other_table.insert().values(bat="hoho"))

    conn.commit()

with engine.begin() as conn:
    conn.execute(some_table.insert().values(foo="bar"))

# 1.4 / 2.0 code

from sqlalchemy import create_engine

engine = create_engine(..., future=True)

with engine.connect() as conn:
    conn.execute(some_table.insert().values(foo="bar"))
    conn.commit()

    conn.execute(text("some other SQL"))
    conn.rollback()

from sqlalchemy import MetaData

metadata_obj = MetaData(bind=engine)  # no longer supported

metadata_obj.create_all()  # requires Engine or Connection

metadata_obj.reflect()  # requires Engine or Connection

t = Table("t", metadata_obj, autoload=True)  # use autoload_with=engine

result = engine.execute(t.select())  # no longer supported

result = t.select().execute()  # no longer supported

from sqlalchemy import MetaData

metadata_obj = MetaData()

# engine level:

# create tables
metadata_obj.create_all(engine)

# reflect all tables
metadata_obj.reflect(engine)

# reflect individual table
t = Table("t", metadata_obj, autoload_with=engine)

# connection level:

with engine.connect() as connection:
    # create tables, requires explicit begin and/or commit:
    with connection.begin():
        metadata_obj.create_all(connection)

    # reflect all tables
    metadata_obj.reflect(connection)

    # reflect individual table
    t = Table("t", metadata_obj, autoload_with=connection)

    # execute SQL statements
    result = connection.execute(t.select())

result = engine.execute(some_statement)

result = some_statement.execute()

# many choices

# bound metadata?
metadata_obj = MetaData(engine)

# or not?
metadata_obj = MetaData()

# execute from engine?
result = engine.execute(stmt)

# or execute the statement itself (but only if you did
# "bound metadata" above, which means you can't get rid of "bound" if any
# part of your program uses this form)
result = stmt.execute()

# execute from connection, but it autocommits?
conn = engine.connect()
conn.execute(stmt)

# execute from connection, but autocommit isn't working, so use the special
# option?
conn.execution_options(autocommit=True).execute(stmt)

# or on the statement ?!
conn.execute(stmt.execution_options(autocommit=True))

# or execute from connection, and we use explicit transaction?
with conn.begin():
    conn.execute(stmt)

# one choice - work with explicit connection, explicit transaction
# (there remain a few variants on how to demarcate the transaction)

# "begin once" - one transaction only per checkout
with engine.begin() as conn:
    result = conn.execute(stmt)

# "commit as you go" - zero or more commits per checkout
with engine.connect() as conn:
    result = conn.execute(stmt)
    conn.commit()

# "commit as you go" but with a transaction block instead of autobegin
with engine.connect() as conn:
    with conn.begin():
        result = conn.execute(stmt)

connection = engine.connect()

# direct string SQL not supported; use text() or exec_driver_sql() method
result = connection.execute("select * from table")

# positional parameters no longer supported, only named
# unless using exec_driver_sql()
result = connection.execute(table.insert(), ("x", "y", "z"))

# **kwargs no longer accepted, pass a single dictionary
result = connection.execute(table.insert(), x=10, y=5)

# multiple *args no longer accepted, pass a list
result = connection.execute(
    table.insert(), {"x": 10, "y": 5}, {"x": 15, "y": 12}, {"x": 9, "y": 8}
)

connection = engine.connect()

from sqlalchemy import text

result = connection.execute(text("select * from table"))

# pass a single dictionary for single statement execution
result = connection.execute(table.insert(), {"x": 10, "y": 5})

# pass a list of dictionaries for executemany
result = connection.execute(
    table.insert(), [{"x": 10, "y": 5}, {"x": 15, "y": 12}, {"x": 9, "y": 8}]
)

engine = create_engine(..., future=True)  # using future mode

with engine.connect() as conn:
    result = conn.execute(text("select x, y from table"))

    row = result.first()  # suppose the row is (1, 2)

    "x" in row  # evaluates to False, in 1.x / future=False, this would be True

    1 in row  # evaluates to True, in 1.x / future=False, this would be False

from sqlalchemy.future.orm import Session

session = Session(some_engine)

result = session.execute(stmt)
for row in result.mappings():
    print("the user is: %s" % row["User"])

from sqlalchemy.future import select

stmt = select(User, Address).join(User.addresses)

for row in session.execute(stmt).mappings():
    print("the user is: %s the address is: %s" % (row[User], row[Address]))

conn = engine.connect()

# won't autocommit in 2.0
conn.execute(some_table.insert().values(foo="bar"))

conn = engine.connect()

# won't autocommit in 2.0
conn.execute(text("INSERT INTO table (foo) VALUES ('bar')"))

conn = engine.connect()

# won't autocommit in 2.0
conn.execute(text("EXEC my_procedural_thing()").execution_options(autocommit=True))

with engine.begin() as conn:
    conn.execute(some_table.insert().values(foo="bar"))
    conn.execute(some_other_table.insert().values(bat="hoho"))

with engine.connect() as conn:
    with conn.begin():
        conn.execute(some_table.insert().values(foo="bar"))
        conn.execute(some_other_table.insert().values(bat="hoho"))

with engine.begin() as conn:
    conn.execute(text("EXEC my_procedural_thing()"))

with engine.connect() as conn:
    conn.execute(some_table.insert().values(foo="bar"))
    conn.execute(some_other_table.insert().values(bat="hoho"))

    conn.commit()

with engine.begin() as conn:
    conn.execute(some_table.insert().values(foo="bar"))

# 1.4 / 2.0 code

from sqlalchemy import create_engine

engine = create_engine(..., future=True)

with engine.connect() as conn:
    conn.execute(some_table.insert().values(foo="bar"))
    conn.commit()

    conn.execute(text("some other SQL"))
    conn.rollback()

from sqlalchemy import MetaData

metadata_obj = MetaData(bind=engine)  # no longer supported

metadata_obj.create_all()  # requires Engine or Connection

metadata_obj.reflect()  # requires Engine or Connection

t = Table("t", metadata_obj, autoload=True)  # use autoload_with=engine

result = engine.execute(t.select())  # no longer supported

result = t.select().execute()  # no longer supported

from sqlalchemy import MetaData

metadata_obj = MetaData()

# engine level:

# create tables
metadata_obj.create_all(engine)

# reflect all tables
metadata_obj.reflect(engine)

# reflect individual table
t = Table("t", metadata_obj, autoload_with=engine)

# connection level:

with engine.connect() as connection:
    # create tables, requires explicit begin and/or commit:
    with connection.begin():
        metadata_obj.create_all(connection)

    # reflect all tables
    metadata_obj.reflect(connection)

    # reflect individual table
    t = Table("t", metadata_obj, autoload_with=connection)

    # execute SQL statements
    result = connection.execute(t.select())

result = engine.execute(some_statement)

result = some_statement.execute()

# many choices

# bound metadata?
metadata_obj = MetaData(engine)

# or not?
metadata_obj = MetaData()

# execute from engine?
result = engine.execute(stmt)

# or execute the statement itself (but only if you did
# "bound metadata" above, which means you can't get rid of "bound" if any
# part of your program uses this form)
result = stmt.execute()

# execute from connection, but it autocommits?
conn = engine.connect()
conn.execute(stmt)

# execute from connection, but autocommit isn't working, so use the special
# option?
conn.execution_options(autocommit=True).execute(stmt)

# or on the statement ?!
conn.execute(stmt.execution_options(autocommit=True))

# or execute from connection, and we use explicit transaction?
with conn.begin():
    conn.execute(stmt)

# one choice - work with explicit connection, explicit transaction
# (there remain a few variants on how to demarcate the transaction)

# "begin once" - one transaction only per checkout
with engine.begin() as conn:
    result = conn.execute(stmt)

# "commit as you go" - zero or more commits per checkout
with engine.connect() as conn:
    result = conn.execute(stmt)
    conn.commit()

# "commit as you go" but with a transaction block instead of autobegin
with engine.connect() as conn:
    with conn.begin():
        result = conn.execute(stmt)

# many choices

# bound metadata?
metadata_obj = MetaData(engine)

# or not?
metadata_obj = MetaData()

# execute from engine?
result = engine.execute(stmt)

# or execute the statement itself (but only if you did
# "bound metadata" above, which means you can't get rid of "bound" if any
# part of your program uses this form)
result = stmt.execute()

# execute from connection, but it autocommits?
conn = engine.connect()
conn.execute(stmt)

# execute from connection, but autocommit isn't working, so use the special
# option?
conn.execution_options(autocommit=True).execute(stmt)

# or on the statement ?!
conn.execute(stmt.execution_options(autocommit=True))

# or execute from connection, and we use explicit transaction?
with conn.begin():
    conn.execute(stmt)

# one choice - work with explicit connection, explicit transaction
# (there remain a few variants on how to demarcate the transaction)

# "begin once" - one transaction only per checkout
with engine.begin() as conn:
    result = conn.execute(stmt)

# "commit as you go" - zero or more commits per checkout
with engine.connect() as conn:
    result = conn.execute(stmt)
    conn.commit()

# "commit as you go" but with a transaction block instead of autobegin
with engine.connect() as conn:
    with conn.begin():
        result = conn.execute(stmt)

connection = engine.connect()

# direct string SQL not supported; use text() or exec_driver_sql() method
result = connection.execute("select * from table")

# positional parameters no longer supported, only named
# unless using exec_driver_sql()
result = connection.execute(table.insert(), ("x", "y", "z"))

# **kwargs no longer accepted, pass a single dictionary
result = connection.execute(table.insert(), x=10, y=5)

# multiple *args no longer accepted, pass a list
result = connection.execute(
    table.insert(), {"x": 10, "y": 5}, {"x": 15, "y": 12}, {"x": 9, "y": 8}
)

connection = engine.connect()

from sqlalchemy import text

result = connection.execute(text("select * from table"))

# pass a single dictionary for single statement execution
result = connection.execute(table.insert(), {"x": 10, "y": 5})

# pass a list of dictionaries for executemany
result = connection.execute(
    table.insert(), [{"x": 10, "y": 5}, {"x": 15, "y": 12}, {"x": 9, "y": 8}]
)

engine = create_engine(..., future=True)  # using future mode

with engine.connect() as conn:
    result = conn.execute(text("select x, y from table"))

    row = result.first()  # suppose the row is (1, 2)

    "x" in row  # evaluates to False, in 1.x / future=False, this would be True

    1 in row  # evaluates to True, in 1.x / future=False, this would be False

from sqlalchemy.future.orm import Session

session = Session(some_engine)

result = session.execute(stmt)
for row in result.mappings():
    print("the user is: %s" % row["User"])

from sqlalchemy.future import select

stmt = select(User, Address).join(User.addresses)

for row in session.execute(stmt).mappings():
    print("the user is: %s the address is: %s" % (row[User], row[Address]))

# select_from / order_by keywords no longer supported
stmt = select([1], select_from=table, order_by=table.c.id)

# whereclause parameter no longer supported
stmt = select([table.c.x], table.c.id == 5)

# whereclause parameter no longer supported
stmt = table.select(table.c.id == 5)

# list emits a deprecation warning
stmt = select([table.c.x, table.c.y])

# list emits a deprecation warning
case_clause = case(
    [(table.c.x == 5, "five"), (table.c.x == 7, "seven")],
    else_="neither five nor seven",
)

# use generative methods
stmt = select(1).select_from(table).order_by(table.c.id)

# use generative methods
stmt = select(table).where(table.c.id == 5)

# use generative methods
stmt = table.select().where(table.c.id == 5)

# pass columns clause expressions positionally
stmt = select(table.c.x, table.c.y)

# case conditions passed positionally
case_clause = case(
    (table.c.x == 5, "five"), (table.c.x == 7, "seven"), else_="neither five nor seven"
)

# table columns for CREATE TABLE - structural
table = Table("table", metadata_obj, Column("x", Integer), Column("y", Integer))

# columns in a SELECT statement - structural
stmt = select(table.c.x, table.c.y)

# literal elements in an IN clause - data
stmt = stmt.where(table.c.y.in_([1, 2, 3]))

# no longer supported
stmt = insert(table, values={"x": 10, "y": 15}, inline=True)

# no longer supported
stmt = insert(table, values={"x": 10, "y": 15}, returning=[table.c.x])

# no longer supported
stmt = table.delete(table.c.x > 15)

# no longer supported
stmt = table.update(table.c.x < 15, preserve_parameter_order=True).values(
    [(table.c.y, 20), (table.c.x, table.c.y + 10)]
)

# use generative methods, **kwargs OK for values()
stmt = insert(table).values(x=10, y=15).inline()

# use generative methods, dictionary also still  OK for values()
stmt = insert(table).values({"x": 10, "y": 15}).returning(table.c.x)

# use generative methods
stmt = table.delete().where(table.c.x > 15)

# use generative methods, ordered_values() replaces preserve_parameter_order
stmt = (
    table.update()
    .where(
        table.c.x < 15,
    )
    .ordered_values((table.c.y, 20), (table.c.x, table.c.y + 10))
)

# select_from / order_by keywords no longer supported
stmt = select([1], select_from=table, order_by=table.c.id)

# whereclause parameter no longer supported
stmt = select([table.c.x], table.c.id == 5)

# whereclause parameter no longer supported
stmt = table.select(table.c.id == 5)

# list emits a deprecation warning
stmt = select([table.c.x, table.c.y])

# list emits a deprecation warning
case_clause = case(
    [(table.c.x == 5, "five"), (table.c.x == 7, "seven")],
    else_="neither five nor seven",
)

# use generative methods
stmt = select(1).select_from(table).order_by(table.c.id)

# use generative methods
stmt = select(table).where(table.c.id == 5)

# use generative methods
stmt = table.select().where(table.c.id == 5)

# pass columns clause expressions positionally
stmt = select(table.c.x, table.c.y)

# case conditions passed positionally
case_clause = case(
    (table.c.x == 5, "five"), (table.c.x == 7, "seven"), else_="neither five nor seven"
)

# table columns for CREATE TABLE - structural
table = Table("table", metadata_obj, Column("x", Integer), Column("y", Integer))

# columns in a SELECT statement - structural
stmt = select(table.c.x, table.c.y)

# literal elements in an IN clause - data
stmt = stmt.where(table.c.y.in_([1, 2, 3]))

# no longer supported
stmt = insert(table, values={"x": 10, "y": 15}, inline=True)

# no longer supported
stmt = insert(table, values={"x": 10, "y": 15}, returning=[table.c.x])

# no longer supported
stmt = table.delete(table.c.x > 15)

# no longer supported
stmt = table.update(table.c.x < 15, preserve_parameter_order=True).values(
    [(table.c.y, 20), (table.c.x, table.c.y + 10)]
)

# use generative methods, **kwargs OK for values()
stmt = insert(table).values(x=10, y=15).inline()

# use generative methods, dictionary also still  OK for values()
stmt = insert(table).values({"x": 10, "y": 15}).returning(table.c.x)

# use generative methods
stmt = table.delete().where(table.c.x > 15)

# use generative methods, ordered_values() replaces preserve_parameter_order
stmt = (
    table.update()
    .where(
        table.c.x < 15,
    )
    .ordered_values((table.c.y, 20), (table.c.x, table.c.y + 10))
)

from sqlalchemy.ext import declarative_base, declared_attr

from sqlalchemy.orm import declarative_base, declared_attr

from sqlalchemy.orm import mapper

mapper(SomeClass, some_table, properties={"related": relationship(SomeRelatedClass)})

from sqlalchemy.orm import registry

mapper_reg = registry()

mapper_reg.map_imperatively(
    SomeClass, some_table, properties={"related": relationship(SomeRelatedClass)}
)

from sqlalchemy.orm import registry

mapper_reg = registry()

Base = mapper_reg.generate_base()

class SomeRelatedClass(Base):
    __tablename__ = "related"

    # ...

mapper_reg.map_imperatively(
    SomeClass,
    some_table,
    properties={
        "related": relationship(
            "SomeRelatedClass",
            primaryjoin="SomeRelatedClass.related_id == SomeClass.id",
        )
    },
)

from sqlalchemy.ext import declarative_base, declared_attr

from sqlalchemy.orm import declarative_base, declared_attr

from sqlalchemy.orm import mapper

mapper(SomeClass, some_table, properties={"related": relationship(SomeRelatedClass)})

from sqlalchemy.orm import registry

mapper_reg = registry()

mapper_reg.map_imperatively(
    SomeClass, some_table, properties={"related": relationship(SomeRelatedClass)}
)

from sqlalchemy.orm import registry

mapper_reg = registry()

Base = mapper_reg.generate_base()

class SomeRelatedClass(Base):
    __tablename__ = "related"

    # ...

mapper_reg.map_imperatively(
    SomeClass,
    some_table,
    properties={
        "related": relationship(
            "SomeRelatedClass",
            primaryjoin="SomeRelatedClass.related_id == SomeClass.id",
        )
    },
)

session.query(User).get(42)

session.get(User, 42)

session.query(User).all()

session.execute(
  select(User)
).scalars().all()

# or

session.scalars(
  select(User)
).all()

session.query(User).\
  filter_by(name="some user").\
  one()

session.execute(
  select(User).
  filter_by(name="some user")
).scalar_one()

session.query(User).\
  filter_by(name="some user").\
  first()

session.scalars(
  select(User).
  filter_by(name="some user").
  limit(1)
).first()

session.query(User).options(
  joinedload(User.addresses)
).all()

session.scalars(
  select(User).
  options(
    joinedload(User.addresses)
  )
).unique().all()

session.query(User).\
  join(Address).\
  filter(
    Address.email == "e@sa.us"
  ).\
  all()

session.execute(
  select(User).
  join(Address).
  where(
    Address.email == "e@sa.us"
  )
).scalars().all()

session.query(User).\
  from_statement(
    text("select * from users")
  ).\
  all()

session.scalars(
  select(User).
  from_statement(
    text("select * from users")
  )
).all()

session.query(User).\
  join(User.addresses).\
  options(
    contains_eager(User.addresses)
  ).\
  populate_existing().all()

session.execute(
  select(User)
  .join(User.addresses)
  .options(
    contains_eager(User.addresses)
  )
  .execution_options(
      populate_existing=True
  )
).scalars().all()

session.query(User).\
  filter(User.name == "foo").\
  update(
    {"fullname": "Foo Bar"},
    synchronize_session="evaluate"
  )

session.execute(
  update(User)
  .where(User.name == "foo")
  .values(fullname="Foo Bar")
  .execution_options(
    synchronize_session="evaluate"
  )
)

session.query(User).count()

session.scalar(
  select(func.count()).
  select_from(User)
)

# or

session.scalar(
  select(func.count(User.id))
)

session = Session(engine)

# becomes legacy use case
user = session.query(User).filter_by(name="some user").one()

# becomes legacy use case
user = session.query(User).filter_by(name="some user").first()

# becomes legacy use case
user = session.query(User).get(5)

# becomes legacy use case
for user in (
    session.query(User).join(User.addresses).filter(Address.email == "some@email.com")
):
    ...

# becomes legacy use case
users = session.query(User).options(joinedload(User.addresses)).order_by(User.id).all()

# becomes legacy use case
users = session.query(User).from_statement(text("select * from users")).all()

# etc

session = Session(engine)

user = session.execute(select(User).filter_by(name="some user")).scalar_one()

# for first(), no LIMIT is applied automatically; add limit(1) if LIMIT
# is desired on the query
user = (
    session.execute(select(User).filter_by(name="some user").limit(1)).scalars().first()
)

# get() moves to the Session directly
user = session.get(User, 5)

for user in session.execute(
    select(User).join(User.addresses).filter(Address.email == "some@email.case")
).scalars():
    ...

# when using joinedload() against collections, use unique() on the result
users = (
    session.execute(select(User).options(joinedload(User.addresses)).order_by(User.id))
    .unique()
    .all()
)

# select() has ORM-ish methods like from_statement() that only work
# if the statement is against ORM entities
users = (
    session.execute(select(User).from_statement(text("select * from users")))
    .scalars()
    .all()
)

# legacy usage
user_obj = session.query(User).get(5)

# 1.4 / 2.0 cross-compatible use
user_obj = session.get(User, 5)

# string use removed
q = session.query(User).join("addresses")

# string use removed
q = session.query(User).options(joinedload("addresses"))

# string use removed
q = session.query(Address).filter(with_parent(u1, "addresses"))

# compatible with all modern SQLAlchemy versions

q = session.query(User).join(User.addresses)

q = session.query(User).options(joinedload(User.addresses))

q = session.query(Address).filter(with_parent(u1, User.addresses))

# SQLAlchemy 1.4 / 2.0 cross compatible use

stmt = select(User).join(User.addresses)
result = session.execute(stmt)

stmt = select(User).options(joinedload(User.addresses))
result = session.execute(stmt)

stmt = select(Address).where(with_parent(u1, User.addresses))
result = session.execute(stmt)

# chaining removed
q = session.query(User).join("orders", "items", "keywords")

q = session.query(User).join(User.orders).join(Order.items).join(Item.keywords)

# 1.4 / 2.0 cross compatible

stmt = select(User).join(User.orders).join(Order.items).join(Item.keywords)
result = session.execute(stmt)

# join_from can also be helpful
stmt = select(User).join_from(User, Order).join_from(Order, Item, Order.items)
result = session.execute(stmt)

# no longer supported
q = (
    session.query(Node)
    .join("children", aliased=True)
    .filter(Node.name == "some sub child")
    .join("children", from_joinpoint=True, aliased=True)
    .filter(Node.name == "some sub sub child")
)

n1 = aliased(Node)
n2 = aliased(Node)

q = (
    select(Node)
    .join(Node.children.of_type(n1))
    .where(n1.name == "some sub child")
    .join(n1.children.of_type(n2))
    .where(n2.name == "some sub child")
)

# 1.xx code

result = (
    session.query(User)
    .join(User.addresses)
    .distinct()
    .order_by(Address.email_address)
    .all()
)

# 1.4 / 2.0 code

stmt = (
    select(User, Address.email_address)
    .join(User.addresses)
    .distinct()
    .order_by(Address.email_address)
)

result = session.execute(stmt).columns(User).all()

# from_self is removed
q = (
    session.query(User, Address.email_address)
    .join(User.addresses)
    .from_self(User)
    .order_by(Address.email_address)
)

from sqlalchemy.orm import aliased

subq = session.query(User, Address.email_address).join(User.addresses).subquery()

ua = aliased(User, subq)

aa = aliased(Address, subq)

q = session.query(ua, aa).order_by(aa.email_address)

from sqlalchemy.orm import aliased

subq = select(User, Address.email_address).join(User.addresses).subquery()

ua = aliased(User, subq)

aa = aliased(Address, subq)

stmt = select(ua, aa).order_by(aa.email_address)

result = session.execute(stmt)

# 1.4 / 2.0 code

subq = select(User, Address).join(User.addresses).subquery()

ua = aliased(User, subq)
aa = aliased(Address, subq)

stmt = select(ua, aa).order_by(aa.email_address)
result = session.execute(stmt)

SELECT  anon_1.id  AS  anon_1_id,  anon_1.id_1  AS  anon_1_id_1,
  anon_1.user_id  AS  anon_1_user_id,
  anon_1.email_address  AS  anon_1_email_address
FROM  (
  SELECT  "user".id  AS  id,  address.id  AS  id_1,
  address.user_id  AS  user_id,  address.email_address  AS  email_address
  FROM  "user"  JOIN  address  ON  "user".id  =  address.user_id
)  AS  anon_1  ORDER  BY  anon_1.email_address

subquery = session.query(User).filter(User.id == 5).subquery()

user = session.query(User).select_entity_from(subquery).first()

from sqlalchemy.orm import aliased

subquery = session.query(User).filter(User.name.like("%somename%")).subquery()

ua = aliased(User, subquery)

user = session.query(ua).order_by(ua.id).first()

from sqlalchemy.orm import aliased

subquery = select(User).where(User.name.like("%somename%")).subquery()

ua = aliased(User, subquery)

# note that LIMIT 1 is not automatically supplied, if needed
user = session.execute(select(ua).order_by(ua.id).limit(1)).scalars().first()

# In the legacy API, many rows each have the same User primary key, but
# only one User per primary key is returned
users = session.query(User).options(joinedload(User.addresses))

# In the new API, uniquing is available but not implicitly
# enabled
result = session.execute(select(User).options(joinedload(User.addresses)))

# this actually will raise an error to let the user know that
# uniquing should be applied
rows = result.all()

# 1.4 / 2.0 code

stmt = select(User).options(joinedload(User.addresses))

# statement will raise if unique() is not used, due to joinedload()
# of a collection.  in all other cases, unique() is not needed.
# By stating unique() explicitly, confusion over discrepancies between
# number of objects/ rows returned vs. "SELECT COUNT(*)" is resolved
rows = session.execute(stmt).unique().all()

class User(Base):
    __tablename__ = "user"

    posts = relationship(Post, lazy="dynamic")

jack = session.get(User, 5)

# filter Jack's blog posts
posts = session.scalars(jack.posts.statement.where(Post.headline == "this is a post"))

from sqlalchemy.orm import with_parent

jack = session.get(User, 5)

posts = session.scalars(
    select(Post)
    .where(with_parent(jack, User.posts))
    .where(Post.headline == "this is a post")
)

from sqlalchemy.orm import Session

sess = Session(engine, autocommit=True)

# no transaction begun, but emits SQL, won't be supported
obj = sess.query(Class).first()

# session flushes in a transaction that it begins and
# commits, won't be supported
sess.flush()

from sqlalchemy.orm import Session

sess = Session(engine)

sess.begin()  # begin explicitly; if not called, will autobegin
# when database access is needed

sess.add(obj)

sess.commit()

import contextlib

@contextlib.contextmanager
def transaction(session):
    if not session.in_transaction():
        with session.begin():
            yield
    else:
        yield

# method_a starts a transaction and calls method_b
def method_a(session):
    with transaction(session):
        method_b(session)

# method_b also starts a transaction, but when
# called from method_a participates in the ongoing
# transaction.
def method_b(session):
    with transaction(session):
        session.add(SomeObject("bat", "lala"))

Session = sessionmaker(engine)

# create a Session and call method_a
with Session() as session:
    method_a(session)

def method_a(session):
    method_b(session)

def method_b(session):
    session.add(SomeObject("bat", "lala"))

Session = sessionmaker(engine)

# create a Session and call method_a
with Session() as session:
    with session.begin():
        method_a(session)

session = Session(engine)

# becomes legacy use case
user = session.query(User).filter_by(name="some user").one()

# becomes legacy use case
user = session.query(User).filter_by(name="some user").first()

# becomes legacy use case
user = session.query(User).get(5)

# becomes legacy use case
for user in (
    session.query(User).join(User.addresses).filter(Address.email == "some@email.com")
):
    ...

# becomes legacy use case
users = session.query(User).options(joinedload(User.addresses)).order_by(User.id).all()

# becomes legacy use case
users = session.query(User).from_statement(text("select * from users")).all()

# etc

session = Session(engine)

user = session.execute(select(User).filter_by(name="some user")).scalar_one()

# for first(), no LIMIT is applied automatically; add limit(1) if LIMIT
# is desired on the query
user = (
    session.execute(select(User).filter_by(name="some user").limit(1)).scalars().first()
)

# get() moves to the Session directly
user = session.get(User, 5)

for user in session.execute(
    select(User).join(User.addresses).filter(Address.email == "some@email.case")
).scalars():
    ...

# when using joinedload() against collections, use unique() on the result
users = (
    session.execute(select(User).options(joinedload(User.addresses)).order_by(User.id))
    .unique()
    .all()
)

# select() has ORM-ish methods like from_statement() that only work
# if the statement is against ORM entities
users = (
    session.execute(select(User).from_statement(text("select * from users")))
    .scalars()
    .all()
)

# legacy usage
user_obj = session.query(User).get(5)

# 1.4 / 2.0 cross-compatible use
user_obj = session.get(User, 5)

# string use removed
q = session.query(User).join("addresses")

# string use removed
q = session.query(User).options(joinedload("addresses"))

# string use removed
q = session.query(Address).filter(with_parent(u1, "addresses"))

# compatible with all modern SQLAlchemy versions

q = session.query(User).join(User.addresses)

q = session.query(User).options(joinedload(User.addresses))

q = session.query(Address).filter(with_parent(u1, User.addresses))

# SQLAlchemy 1.4 / 2.0 cross compatible use

stmt = select(User).join(User.addresses)
result = session.execute(stmt)

stmt = select(User).options(joinedload(User.addresses))
result = session.execute(stmt)

stmt = select(Address).where(with_parent(u1, User.addresses))
result = session.execute(stmt)

# chaining removed
q = session.query(User).join("orders", "items", "keywords")

q = session.query(User).join(User.orders).join(Order.items).join(Item.keywords)

# 1.4 / 2.0 cross compatible

stmt = select(User).join(User.orders).join(Order.items).join(Item.keywords)
result = session.execute(stmt)

# join_from can also be helpful
stmt = select(User).join_from(User, Order).join_from(Order, Item, Order.items)
result = session.execute(stmt)

# no longer supported
q = (
    session.query(Node)
    .join("children", aliased=True)
    .filter(Node.name == "some sub child")
    .join("children", from_joinpoint=True, aliased=True)
    .filter(Node.name == "some sub sub child")
)

n1 = aliased(Node)
n2 = aliased(Node)

q = (
    select(Node)
    .join(Node.children.of_type(n1))
    .where(n1.name == "some sub child")
    .join(n1.children.of_type(n2))
    .where(n2.name == "some sub child")
)

# 1.xx code

result = (
    session.query(User)
    .join(User.addresses)
    .distinct()
    .order_by(Address.email_address)
    .all()
)

# 1.4 / 2.0 code

stmt = (
    select(User, Address.email_address)
    .join(User.addresses)
    .distinct()
    .order_by(Address.email_address)
)

result = session.execute(stmt).columns(User).all()

# from_self is removed
q = (
    session.query(User, Address.email_address)
    .join(User.addresses)
    .from_self(User)
    .order_by(Address.email_address)
)

from sqlalchemy.orm import aliased

subq = session.query(User, Address.email_address).join(User.addresses).subquery()

ua = aliased(User, subq)

aa = aliased(Address, subq)

q = session.query(ua, aa).order_by(aa.email_address)

from sqlalchemy.orm import aliased

subq = select(User, Address.email_address).join(User.addresses).subquery()

ua = aliased(User, subq)

aa = aliased(Address, subq)

stmt = select(ua, aa).order_by(aa.email_address)

result = session.execute(stmt)

# 1.4 / 2.0 code

subq = select(User, Address).join(User.addresses).subquery()

ua = aliased(User, subq)
aa = aliased(Address, subq)

stmt = select(ua, aa).order_by(aa.email_address)
result = session.execute(stmt)

SELECT  anon_1.id  AS  anon_1_id,  anon_1.id_1  AS  anon_1_id_1,
  anon_1.user_id  AS  anon_1_user_id,
  anon_1.email_address  AS  anon_1_email_address
FROM  (
  SELECT  "user".id  AS  id,  address.id  AS  id_1,
  address.user_id  AS  user_id,  address.email_address  AS  email_address
  FROM  "user"  JOIN  address  ON  "user".id  =  address.user_id
)  AS  anon_1  ORDER  BY  anon_1.email_address

subquery = session.query(User).filter(User.id == 5).subquery()

user = session.query(User).select_entity_from(subquery).first()

from sqlalchemy.orm import aliased

subquery = session.query(User).filter(User.name.like("%somename%")).subquery()

ua = aliased(User, subquery)

user = session.query(ua).order_by(ua.id).first()

from sqlalchemy.orm import aliased

subquery = select(User).where(User.name.like("%somename%")).subquery()

ua = aliased(User, subquery)

# note that LIMIT 1 is not automatically supplied, if needed
user = session.execute(select(ua).order_by(ua.id).limit(1)).scalars().first()

# In the legacy API, many rows each have the same User primary key, but
# only one User per primary key is returned
users = session.query(User).options(joinedload(User.addresses))

# In the new API, uniquing is available but not implicitly
# enabled
result = session.execute(select(User).options(joinedload(User.addresses)))

# this actually will raise an error to let the user know that
# uniquing should be applied
rows = result.all()

# 1.4 / 2.0 code

stmt = select(User).options(joinedload(User.addresses))

# statement will raise if unique() is not used, due to joinedload()
# of a collection.  in all other cases, unique() is not needed.
# By stating unique() explicitly, confusion over discrepancies between
# number of objects/ rows returned vs. "SELECT COUNT(*)" is resolved
rows = session.execute(stmt).unique().all()

class User(Base):
    __tablename__ = "user"

    posts = relationship(Post, lazy="dynamic")

jack = session.get(User, 5)

# filter Jack's blog posts
posts = session.scalars(jack.posts.statement.where(Post.headline == "this is a post"))

from sqlalchemy.orm import with_parent

jack = session.get(User, 5)

posts = session.scalars(
    select(Post)
    .where(with_parent(jack, User.posts))
    .where(Post.headline == "this is a post")
)

from sqlalchemy.orm import Session

sess = Session(engine, autocommit=True)

# no transaction begun, but emits SQL, won't be supported
obj = sess.query(Class).first()

# session flushes in a transaction that it begins and
# commits, won't be supported
sess.flush()

from sqlalchemy.orm import Session

sess = Session(engine)

sess.begin()  # begin explicitly; if not called, will autobegin
# when database access is needed

sess.add(obj)

sess.commit()

import contextlib

@contextlib.contextmanager
def transaction(session):
    if not session.in_transaction():
        with session.begin():
            yield
    else:
        yield

# method_a starts a transaction and calls method_b
def method_a(session):
    with transaction(session):
        method_b(session)

# method_b also starts a transaction, but when
# called from method_a participates in the ongoing
# transaction.
def method_b(session):
    with transaction(session):
        session.add(SomeObject("bat", "lala"))

Session = sessionmaker(engine)

# create a Session and call method_a
with Session() as session:
    method_a(session)

def method_a(session):
    method_b(session)

def method_b(session):
    session.add(SomeObject("bat", "lala"))

Session = sessionmaker(engine)

# create a Session and call method_a
with Session() as session:
    with session.begin():
        method_a(session)