>>> obj = MyObj()
>>> obj.some_value
None

>>> obj = MyObj()
>>> obj.some_value
"my default"

result = session.query(
    func.substr(A.some_thing, 0, 4), A
).options(joinedload(A.bs)).all()

users = session.query(
    func.date(
        User.date_created, 'start of month'
    ).label('month'),
    User,
).options(joinedload(User.orders)).all()

result = session.query(
    func.substr(A.some_thing, 0, 4, type_=String), A
).options(joinedload(A.bs)).all()

users = session.query(
    func.date(
        User.date_created, 'start of month', type_=DateTime
    ).label('month'),
    User,
).options(joinedload(User.orders)).all()

>>> some_user = User()
>>> q = s.query(User).filter(User.name == some_user)
...
sqlalchemy.exc.ArgumentError: Object <__main__.User object at 0x103167e90> is not legal as a SQL literal value

>>> # Address.user refers to the User mapper, so
>>> # this is of course still OK!
>>> q = s.query(Address).filter(Address.user == some_user)

class Person(Base):
    __tablename__ = 'person'

    id = Column(Integer, primary_key=True)
    data = Column(JSON)

    name = index_property('data', 'name')

>>> person = Person(name='foobar')
>>> person.name
foobar

class Widget(Base):
    __tablename__ = 'widget'
    id = Column(Integer, primary_key=True)
    type = Column(String)
    data = Column(String)
    __mapper_args__ = {'polymorphic_on': type}


class FooWidget(Widget):
    __mapper_args__ = {'polymorphic_identity': 'foo'}

q = session.query(FooWidget).filter(FooWidget.data == 'bar').exists()

session.query(q).all()

SELECT EXISTS (SELECT 1
FROM widget
WHERE widget.data = :data_1 AND widget.type IN (:type_1)) AS anon_1

s = Session()
s.begin_nested()

s.add(SomeObject())

try:
    # assume the flush fails, flush goes to rollback to the
    # savepoint and that also fails
    s.flush()
except Exception as err:
    print("Something broke, and our SAVEPOINT vanished too")

# this is the SAVEPOINT transaction, marked as
# DEACTIVE so the rollback() call succeeds
s.rollback()

# this is the outermost transaction, remains ACTIVE
# so rollback() or commit() can succeed
s.rollback()

from sqlalchemy import Column, create_engine
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class A(Base):
    __tablename__ = 'a'
    id = Column(Integer, primary_key=True)

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

s = Session(e)

# persist an object
s.add(A(id=1))
s.flush()

# rollback buffer loses reference to A

# load it again, rollback buffer knows nothing
# about it
a1 = s.query(A).first()

# roll back the transaction; all state is expired but the
# "a1" reference remains
s.rollback()

# previous "a1" conflicts with the new one because we aren't
# checking that it never got committed
s.add(A(id=1))
s.commit()

FlushError: New instance <User at 0x7f0287eca4d0> with identity key
(<class 'test.orm.test_transaction.User'>, ('u1',)) conflicts
with persistent instance <User at 0x7f02889c70d0>

from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class A(Base):
    __tablename__ = "a"
    id = Column('id', Integer, primary_key=True)
    type = Column(String)

    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': 'a',
        'passive_deletes': True
    }


class B(A):
    __tablename__ = 'b'
    b_table_id = Column('b_table_id', Integer, primary_key=True)
    bid = Column('bid', Integer, ForeignKey('a.id', ondelete="CASCADE"))
    data = Column('data', String)

    __mapper_args__ = {
        'polymorphic_identity': 'b'
    }

session.delete(some_b)
session.commit()

DELETE FROM a WHERE a.id = %(id)s
{'id': 1}
COMMIT

class A(Base):
    __tablename__ = 'a'
    id = Column(Integer, primary_key=True)
    b = relationship("B", foreign_keys="B.a_id", backref="a")

class A1(A):
    __tablename__ = 'a1'
    id = Column(Integer, primary_key=True)
    b = relationship("B", foreign_keys="B.a1_id", backref="a1")
    __mapper_args__ = {'concrete': True}

class B(Base):
    __tablename__ = 'b'
    id = Column(Integer, primary_key=True)

    a_id = Column(ForeignKey('a.id'))
    a1_id = Column(ForeignKey('a1.id'))

class A(Base):
    __tablename__ = 'a'
    id = Column(Integer, primary_key=True)


class A1(A):
    __tablename__ = 'a1'
    id = Column(Integer, primary_key=True)
    __mapper_args__ = {'concrete': True}


class B(Base):
    __tablename__ = 'b'
    id = Column(Integer, primary_key=True)

    a_id = Column(ForeignKey('a.id'))
    a1_id = Column(ForeignKey('a1.id'))

    a = relationship("A", backref="b")
    a1 = relationship("A1", backref="b")

class A(Base):
    __tablename__ = 'a'
    id = Column(Integer, primary_key=True)
    bs = relationship("B")


class ASub(A):
    __tablename__ = 'a_sub'
    id = Column(Integer, ForeignKey('a.id'), primary_key=True)
    bs = relationship("B")


class B(Base):
    __tablename__ = 'b'
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey('a.id'))

class A(Base):
    __tablename__ = 'a'
    id = Column(Integer, primary_key=True)

    name = Column(String)

    @hybrid_property
    def some_name(self):
        """The name field"""
        return self.name

>>> A.some_name.__doc__
The name field

>>> assert A.name is A.some_name

>>> A.some_name
<sqlalchemy.orm.attributes.hybrid_propertyProxy object at 0x7fde03888230>

>>> A.some_name.info['foo'] = 'bar'
>>> from sqlalchemy import inspect
>>> inspect(A).all_orm_descriptors['some_name'].info
{'foo': 'bar'}

u1 = User(id=7, name='x')
u1.orders = [
    Order(description='o1', address=Address(id=1, email_address='a')),
    Order(description='o2', address=Address(id=1, email_address='b')),
    Order(description='o3', address=Address(id=1, email_address='c'))
]

sess = Session()
sess.merge(u1)

some_object = SomeClass()
session.add(some_object)
some_object.parent_id = some_parent.id
some_object.parent = some_parent

# before the fix
assert some_object not in some_parent.items

# after the fix, backref fired off for some_object.parent = some_parent
assert some_object in some_parent.items

sess.query(Person.name)
            .filter(
                sess.query(Company.name).
                filter(Company.company_id == Person.company_id).
                correlate(Person).as_scalar() == "Elbonia, Inc.")

SELECT people.name AS people_name
FROM people
LEFT OUTER JOIN engineers ON people.person_id = engineers.person_id
LEFT OUTER JOIN managers ON people.person_id = managers.person_id
WHERE (SELECT companies.name
FROM companies
WHERE companies.company_id = people.company_id) = ?

-- old, incorrect query
SELECT people.name AS people_name
FROM people
LEFT OUTER JOIN engineers ON people.person_id = engineers.person_id
LEFT OUTER JOIN managers ON people.person_id = managers.person_id
WHERE (SELECT companies.name
FROM companies, people
WHERE companies.company_id = people.company_id) = ?

# works with all SQLAlchemy versions and all types of polymorphic
# aliasing.

paliased = aliased(Person)
sess.query(paliased.name)
            .filter(
                sess.query(Company.name).
                filter(Company.company_id == paliased.company_id).
                correlate(paliased).as_scalar() == "Elbonia, Inc.")

class A(Base):
    __tablename__ = 'a'
    id = Column(Integer, primary_key=True)
    b_id = Column(ForeignKey('b.id'))
    c_id = Column(ForeignKey('c.id'))

    b = relationship("B")
    c = relationship("C")


class B(Base):
    __tablename__ = 'b'
    id = Column(Integer, primary_key=True)
    c_id = Column(ForeignKey('c.id'))

    c = relationship("C")


class C(Base):
    __tablename__ = 'c'
    id = Column(Integer, primary_key=True)
    d_id = Column(ForeignKey('d.id'))
    d = relationship("D")


class D(Base):
    __tablename__ = 'd'
    id = Column(Integer, primary_key=True)


c_alias_1 = aliased(C)
c_alias_2 = aliased(C)

q = s.query(A)
q = q.join(A.b).join(c_alias_1, B.c).join(c_alias_1.d)
q = q.options(contains_eager(A.b).contains_eager(B.c, alias=c_alias_1).contains_eager(C.d))
q = q.join(c_alias_2, A.c)
q = q.options(contains_eager(A.c, alias=c_alias_2))

SELECT
    d.id AS d_id,
    c_1.id AS c_1_id, c_1.d_id AS c_1_d_id,
    b.id AS b_id, b.c_id AS b_c_id,
    c_2.id AS c_2_id, c_2.d_id AS c_2_d_id,
    a.id AS a_id, a.b_id AS a_b_id, a.c_id AS a_c_id
FROM
    a
    JOIN b ON b.id = a.b_id
    JOIN c AS c_1 ON c_1.id = b.c_id
    JOIN d ON d.id = c_1.d_id
    JOIN c AS c_2 ON c_2.id = a.c_id

>>> from sqlalchemy.orm import raiseload
>>> a1 = s.query(A).options(raiseload(A.some_b)).first()
>>> a1.some_b
Traceback (most recent call last):
...
sqlalchemy.exc.InvalidRequestError: 'A.some_b' is not available due to lazy='raise'

>>> from sqlalchemy.orm import raiseload
>>> a1 = s.query(A).options(raiseload(A.some_b, sql_only=True)).first()
>>> a1.some_b
Traceback (most recent call last):
...
sqlalchemy.exc.InvalidRequestError: 'A.bs' is not available due to lazy='raise_on_sql'

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

@event.listens_for(engine, "handle_error")
def cancel_disconnect(ctx):
    if isinstance(ctx.original_exception, KeyboardInterrupt):
        ctx.is_disconnect = False

>>> from sqlalchemy import table, column, select, literal, exists
>>> orders = table(
...     'orders',
...     column('region'),
...     column('amount'),
...     column('product'),
...     column('quantity')
... )
>>>
>>> upsert = (
...     orders.update()
...     .where(orders.c.region == 'Region1')
...     .values(amount=1.0, product='Product1', quantity=1)
...     .returning(*(orders.c._all_columns)).cte('upsert'))
>>>
>>> insert = orders.insert().from_select(
...     orders.c.keys(),
...     select([
...         literal('Region1'), literal(1.0),
...         literal('Product1'), literal(1)
...     ]).where(~exists(upsert.select()))
... )
>>>
>>> print(insert)  # note formatting added for clarity
WITH upsert AS
(UPDATE orders SET amount=:amount, product=:product, quantity=:quantity
 WHERE orders.region = :region_1
 RETURNING orders.region, orders.amount, orders.product, orders.quantity
)
INSERT INTO orders (region, amount, product, quantity)
SELECT
    :param_1 AS anon_1, :param_2 AS anon_2,
    :param_3 AS anon_3, :param_4 AS anon_4
WHERE NOT (
    EXISTS (
        SELECT upsert.region, upsert.amount,
               upsert.product, upsert.quantity
        FROM upsert))

>>> from sqlalchemy import func

>>> print(func.row_number().over(order_by='x', range_=(-5, 10)))
row_number() OVER (ORDER BY x RANGE BETWEEN :param_1 PRECEDING AND :param_2 FOLLOWING)

>>> print(func.row_number().over(order_by='x', rows=(None, 0)))
row_number() OVER (ORDER BY x ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

>>> print(func.row_number().over(order_by='x', range_=(-2, None)))
row_number() OVER (ORDER BY x RANGE BETWEEN :param_1 PRECEDING AND UNBOUNDED FOLLOWING)

>>> from sqlalchemy import table, column, select, true
>>> people = table('people', column('people_id'), column('age'), column('name'))
>>> books = table('books', column('book_id'), column('owner_id'))
>>> subq = select([books.c.book_id]).\
...      where(books.c.owner_id == people.c.people_id).lateral("book_subq")
>>> print(select([people]).select_from(people.join(subq, true())))
SELECT people.people_id, people.age, people.name
FROM people JOIN LATERAL (SELECT books.book_id AS book_id
FROM books WHERE books.owner_id = people.people_id)
AS book_subq ON true

from sqlalchemy import func

selectable = people.tablesample(
            func.bernoulli(1),
            name='alias',
            seed=func.random())
stmt = select([selectable.c.people_id])

SELECT alias.people_id FROM
people AS alias TABLESAMPLE bernoulli(:bernoulli_1)
REPEATABLE (random())

Table(
    'some_table', metadata,
    Column('x', Integer, primary_key=True),
    Column('y', Integer, primary_key=True)
)

# old way
Table(
    'some_table', metadata,
    Column('x', Integer, primary_key=True, autoincrement=False),
    Column('y', Integer, primary_key=True, autoincrement=False)
)

# column 'y' will be SERIAL/AUTO_INCREMENT/ auto-generating
Table(
    'some_table', metadata,
    Column('x', Integer, primary_key=True),
    Column('y', Integer, primary_key=True, autoincrement=True)
)

Table(
    'b', metadata,
    Column('x', Integer, primary_key=True),
    Column('y', Integer, primary_key=True)
)

SAWarning: Column 'b.x' is marked as a member of the primary
key for table 'b', but has no Python-side or server-side default
generator indicated, nor does it indicate 'autoincrement=True',
and no explicit value is passed.  Primary key columns may not
store NULL. Note that as of SQLAlchemy 1.1, 'autoincrement=True'
must be indicated explicitly for composite (e.g. multicolumn)
primary keys if AUTO_INCREMENT/SERIAL/IDENTITY behavior is
expected for one of the columns in the primary key. CREATE TABLE
statements are impacted by this change as well on most backends.

Table(
    'b', metadata,
    Column('x', Integer, primary_key=True, server_default=FetchedValue()),
    Column('y', Integer, primary_key=True, server_default=FetchedValue())
)

Table(
    'b', metadata,
    Column('x', Integer, primary_key=True),
    Column('y', Integer, primary_key=True, nullable=True)
)

>>> print(column('x').is_distinct_from(None))
x IS DISTINCT FROM NULL

>>> print(column('x').isnot_distinct_from(False))
x IS NOT DISTINCT FROM false

>>> from sqlalchemy.dialects import sqlite
>>> print(column('x').is_distinct_from(None).compile(dialect=sqlite.dialect()))
x IS NOT NULL

stmt = select([t1]).select_from(t1.outerjoin(t2, full=True))

q = session.query(MyClass).outerjoin(MyOtherClass, full=True)

import enum
from sqlalchemy import Table, MetaData, Column, Enum, create_engine


class MyEnum(enum.Enum):
    one = 1
    two = 2
    three = 3


t = Table(
    'data', MetaData(),
    Column('value', Enum(MyEnum))
)

e = create_engine("sqlite://")
t.create(e)

e.execute(t.insert(), {"value": MyEnum.two})
assert e.scalar(t.select()) is MyEnum.two

>>> from sqlalchemy import create_engine
>>> e = create_engine("sqlite://")
>>> row = e.execute("select 1, 2, 3").first()
>>> row[-1], row[-2], row[1], row[-2:2]
3 2 2 (2,)

>>> from sqlalchemy import Table, MetaData, Column, Enum, create_engine
>>> t = Table(
...     'data', MetaData(),
...     Column('value', Enum("one", "two", "three", validate_strings=True))
... )
>>> e = create_engine("sqlite://")
>>> t.create(e)
>>> e.execute(t.insert(), {"value": "four"})
Traceback (most recent call last):
  ...
sqlalchemy.exc.StatementError: (exceptions.LookupError)
"four" is not among the defined enum values
[SQL: u'INSERT INTO data (value) VALUES (?)']
[parameters: [{'value': 'four'}]]

>>> from sqlalchemy import create_engine
>>> import random
>>> e = create_engine("sqlite://", echo='debug')
>>> some_value = ''.join(chr(random.randint(52, 85)) for i in range(5000))
>>> row = e.execute("select ?", [some_value]).first()
... (lines are wrapped for clarity) ...
2016-02-17 13:23:03,027 INFO sqlalchemy.engine.base.Engine select ?
2016-02-17 13:23:03,027 INFO sqlalchemy.engine.base.Engine
('E6@?>9HPOJB<<BHR:@=TS:5ILU=;JLM<4?B9<S48PTNG9>:=TSTLA;9K;9FPM4M8M@;NM6GU
LUAEBT9QGHNHTHR5EP75@OER4?SKC;D:TFUMD:M>;C6U:JLM6R67GEK<A6@S@C@J7>4=4:P
GJ7HQ6 ... (4702 characters truncated) ... J6IK546AJMB4N6S9L;;9AKI;=RJP
HDSSOTNBUEEC9@Q:RCL:I@5?FO<9K>KJAGAO@E6@A7JI8O:J7B69T6<8;F:S;4BEIJS9HM
K:;5OLPM@JR;R:J6<SOTTT=>Q>7T@I::OTDC:CC<=NGP6C>BC8N',)
2016-02-17 13:23:03,027 DEBUG sqlalchemy.engine.base.Engine Col ('?',)
2016-02-17 13:23:03,027 DEBUG sqlalchemy.engine.base.Engine
Row (u'E6@?>9HPOJB<<BHR:@=TS:5ILU=;JLM<4?B9<S48PTNG9>:=TSTLA;9K;9FPM4M8M@;
NM6GULUAEBT9QGHNHTHR5EP75@OER4?SKC;D:TFUMD:M>;C6U:JLM6R67GEK<A6@S@C@J7
>4=4:PGJ7HQ ... (4703 characters truncated) ... J6IK546AJMB4N6S9L;;9AKI;=
RJPHDSSOTNBUEEC9@Q:RCL:I@5?FO<9K>KJAGAO@E6@A7JI8O:J7B69T6<8;F:S;4BEIJS9HM
K:;5OLPM@JR;R:J6<SOTTT=>Q>7T@I::OTDC:CC<=NGP6C>BC8N',)
>>> print(row)
(u'E6@?>9HPOJB<<BHR:@=TS:5ILU=;JLM<4?B9<S48PTNG9>:=TSTLA;9K;9FPM4M8M@;NM6
GULUAEBT9QGHNHTHR5EP75@OER4?SKC;D:TFUMD:M>;C6U:JLM6R67GEK<A6@S@C@J7>4
=4:PGJ7HQ ... (4703 characters truncated) ... J6IK546AJMB4N6S9L;;9AKI;
=RJPHDSSOTNBUEEC9@Q:RCL:I@5?FO<9K>KJAGAO@E6@A7JI8O:J7B69T6<8;F:S;4BEIJS9H
MK:;5OLPM@JR;R:J6<SOTTT=>Q>7T@I::OTDC:CC<=NGP6C>BC8N',)

mytable = Table("mytable", metadata,
        Column("data", ARRAY(Integer, dimensions=2))
    )

expr = mytable.c.data[5][6]

expr = mytable.c.data[5].any(12)

from sqlalchemy import any_, all_

select([mytable]).where(12 == any_(mytable.c.data[5]))

from sqlalchemy import any_, all_

subq = select([mytable.c.value])
select([mytable]).where(12 > any_(subq))

from sqlalchemy import func
stmt = select([func.array_agg(table.c.value)])

from sqlalchemy.dialects.postgresql import aggregate_order_by
expr = func.array_agg(aggregate_order_by(table.c.a, table.c.b.desc()))
stmt = select([expr])

SELECT array_agg(table1.a ORDER BY table1.b DESC) AS array_agg_1 FROM table1

from sqlalchemy.dialects.postgresql import array_agg
stmt = select([array_agg(table.c.value).contains('foo')])

from sqlalchemy import func
stmt = select([
    department.c.id,
    func.percentile_cont(0.5).within_group(
        department.c.salary.desc()
    )
])

SELECT department.id, percentile_cont(0.5)
WITHIN GROUP (ORDER BY department.salary DESC)

# old way
class MyEnum(TypeDecorator, SchemaType):
    impl = postgresql.ENUM('one', 'two', 'three', name='myenum')

    def _set_table(self, table):
        self.impl._set_table(table)

# new way
class MyEnum(TypeDecorator):
    impl = postgresql.ENUM('one', 'two', 'three', name='myenum')

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)

    __table_args__ = {'schema': 'per_user'}

session = Session()
session.connection(execution_options={
    "schema_translate_map": {"per_user": "account_one"}})

# will query from the ``account_one.user`` table
session.query(User).get(5)

>>> from sqlalchemy import table, column
t>>> t = table('x', column('a'), column('b'))
>>> print(t.insert().returning(t.c.a, t.c.b))
INSERT INTO x (a, b) VALUES (:a, :b) RETURNING x.a, x.b

class StringAsInt(TypeDecorator):
    impl = String

    def column_expression(self, col):
        return cast(col, Integer)

    def bind_expression(self, value):
        return cast(value, String)

class Person(Base):
    __tablename__ = 'person'
    id = Column(StringAsInt, primary_key=True)

    pets = relationship(
        'Pets',
        primaryjoin=(
            'foreign(Pets.person_id)'
            '==cast(type_coerce(Person.id, Integer), Integer)'
        )
    )

class Pets(Base):
    __tablename__ = 'pets'
    id = Column('id', Integer, primary_key=True)
    person_id = Column('person_id', Integer)

SELECT person.id AS person_id, pets_1.id AS pets_1_id,
       pets_1.person_id AS pets_1_person_id
FROM person
LEFT OUTER JOIN pets AS pets_1
ON pets_1.person_id = CAST(person.id AS INTEGER)

SELECT pets.id AS pets_id, pets.person_id AS pets_person_id
FROM pets
WHERE pets.person_id = CAST(CAST(%(param_1)s AS VARCHAR) AS INTEGER)
{'param_1': 5}

SELECT pets.id AS pets_id, pets.person_id AS pets_person_id
FROM pets
WHERE pets.person_id = CAST(%(param_1)s AS INTEGER)
{'param_1': 5}

class MyObject(Base):
    # ...

    json_value = Column(JSON(none_as_null=False), nullable=False)

obj = MyObject()  # note no json_value
session.add(obj)
session.commit()  # will fail with integrity error

class MyObject(Base):
    # ...

    json_value = Column(
        JSON(none_as_null=False), nullable=False, default=JSON.NULL)

obj = MyObject(json_value=None)
session.add(obj)
session.commit()  # will insert JSON NULL

# default=None is the same as omitting it entirely, does not apply JSON NULL
json_value = Column(JSON(none_as_null=False), nullable=False, default=None)

q = session.query(User.id, User.name.label('name')).\
    distinct().\
    order_by(User.id, User.name, User.fullname)

SELECT DISTINCT user.id AS a_id, user.name AS name,
 user.fullname AS a_fullname
FROM a ORDER BY user.id, user.name, user.fullname

SELECT DISTINCT user.id AS a_id, user.name AS name, user.name AS a_name,
  user.fullname AS a_fullname
FROM a ORDER BY user.id, user.name, user.fullname

class A(Base):
    __tablename__ = 'a'
    id = Column(Integer, primary_key=True)

    data = Column(String)

    @validates("data")
    def _validate_data_one(self):
        assert "x" in data

    @validates("data")
    def _validate_data_two(self):
        assert "y" in data

configure_mappers()

sqlalchemy.exc.InvalidRequestError: A validation function for mapped attribute 'data' on mapper Mapper|A|a already exists.

stmt = text("SELECT id, name, description FROM table")

# no longer matches by name
stmt = stmt.columns(my_table.c.name, my_table.c.description, my_table.c.id)

# correct version
stmt = stmt.columns(my_table.c.id, my_table.c.name, my_table.c.description)

stmt = text("SELECT * FROM table")
stmt = stmt.columns(my_table.c.id, my_table.c.name, my_table.c.description)

>>> from sqlalchemy.schema import MetaData, Table, Column, CreateTable
>>> from sqlalchemy.types import String
>>> t = Table('t', MetaData(), Column('x', String(), server_default="hi ' there"))
>>> print(CreateTable(t))

CREATE TABLE t (
    x VARCHAR DEFAULT 'hi '' there'
)

(SELECT x FROM table1 ORDER BY y LIMIT 1) UNION
(SELECT x FROM table2 ORDER BY y LIMIT 2)

stmt1 = select([table1.c.x]).order_by(table1.c.y).limit(1)
stmt2 = select([table1.c.x]).order_by(table2.c.y).limit(2)

stmt = union(stmt1, stmt2)

stmt1 = select([table1.c.x]).order_by(table1.c.y).limit(1).alias().select()
stmt2 = select([table2.c.x]).order_by(table2.c.y).limit(2).alias().select()

stmt = union(stmt1, stmt2)

stmt1 = session.query(Model1).order_by(Model1.y).limit(1).subquery().select()
stmt2 = session.query(Model2).order_by(Model2.y).limit(1).subquery().select()

stmt = session.query(Model1).from_statement(stmt1.union(stmt2))

from sqlalchemy.dialects.postgresql import insert

insert_stmt = insert(my_table). \\
    values(id='some_id', data='some data to insert')

do_update_stmt = insert_stmt.on_conflict_do_update(
    index_elements=[my_table.c.id],
    set_=dict(data='some data to update')
)

conn.execute(do_update_stmt)

INSERT INTO my_table (id, data)
VALUES (:id, :data)
ON CONFLICT id DO UPDATE SET data=:data_2

expr = json_col['somekey'].cast(Integer)

expr = json_col['somekey'].astext.cast(Integer)

enum = Enum(
    'manager', 'place_admin', 'carwash_admin',
    'parking_admin', 'service_admin', 'tire_admin',
    'mechanic', 'carwasher', 'tire_mechanic', name="work_place_roles")

class WorkPlacement(Base):
    __tablename__ = 'work_placement'
    id = Column(Integer, primary_key=True)
    roles = Column(ARRAY(enum))


e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
Base.metadata.create_all(e)

CREATE TYPE work_place_roles AS ENUM (
    'manager', 'place_admin', 'carwash_admin', 'parking_admin',
    'service_admin', 'tire_admin', 'mechanic', 'carwasher',
    'tire_mechanic')

CREATE TABLE work_placement (
    id SERIAL NOT NULL,
    roles work_place_roles[],
    PRIMARY KEY (id)
)

from sqlalchemy import inspect
insp = inspect(engine)

plain_views = insp.get_view_names(include='plain')
all_views = insp.get_view_names(include=('plain', 'materialized'))

connection = engine.connect()
connection = connection.execution_options(
    isolation_level="AUTOCOMMIT"
)

t = Table(
    'some_table', metadata,
    Column('x', Integer, primary_key=True, autoincrement=False),
    Column('y', Integer, primary_key=True, autoincrement=True),
    mysql_engine='InnoDB'
)

CREATE TABLE some_table (
    x INTEGER NOT NULL,
    y INTEGER NOT NULL AUTO_INCREMENT,
    PRIMARY KEY (x, y),
    KEY idx_autoinc_y (y)
)ENGINE=InnoDB

CREATE TABLE some_table (
    x INTEGER NOT NULL,
    y INTEGER NOT NULL AUTO_INCREMENT,
    PRIMARY KEY (y, x)
)ENGINE=InnoDB

t = Table(
    'some_table', metadata,
    Column('x', Integer, primary_key=True),
    Column('y', Integer, primary_key=True, autoincrement=True),
    PrimaryKeyConstraint('x', 'y'),
    UniqueConstraint('y'),
    mysql_engine='InnoDB'
)

t = Table(
    'some_table', metadata,
    Column('x', Integer, primary_key=True),
    Column('y', Integer, primary_key=True, autoincrement=True),
    mysql_engine='InnoDB'
)

engine = create_engine(
    "mssql+pyodbc://scott:tiger@ms_2008",
    isolation_level="REPEATABLE READ"
)

>>> from sqlalchemy import create_engine, inspect
>>> engine = create_engine('mssql+pyodbc://scott:tiger@ms_2008', echo=True)
>>> engine.execute("create table s (x varchar(max), y varbinary(max))")
>>> insp = inspect(engine)
>>> for col in insp.get_columns("s"):
...     print(col['type'].__class__, col['type'].length)
...
<class 'sqlalchemy.sql.sqltypes.VARCHAR'> max
<class 'sqlalchemy.dialects.mssql.base.VARBINARY'> max

>>> for col in insp.get_columns("s"):
...     print(col['type'].__class__, col['type'].length)
...
<class 'sqlalchemy.sql.sqltypes.VARCHAR'> None
<class 'sqlalchemy.dialects.mssql.base.VARBINARY'> None

account_table = Table(
    'account', metadata,
    Column('id', Integer, primary_key=True),
    Column('info', String(100)),
    schema="customer_schema"
)

>>> eng = create_engine("mssql+pymssql://mydsn", legacy_schema_aliasing=True)
>>> print(account_table.select().compile(eng))
SELECT account_1.id, account_1.info
FROM customer_schema.account AS account_1