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())

from sqlalchemy import create_engine

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

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())

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 = conn.execute(t.select())

result = engine.execute(some_statement)

result = some_statement.execute()

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 = 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()

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()