>>> class User(Base):
...     __tablename__ = 'user'
...     id = Column(Integer, primary_key=True)
...     name = Column(String)
...     name_syn = synonym(name)
...     addresses = relationship("Address")
...

>>> # universal entry point is inspect()
>>> b = inspect(User)

>>> # b in this case is the Mapper
>>> b
<Mapper at 0x101521950; User>

>>> # Column namespace
>>> b.columns.id
Column('id', Integer(), table=<user>, primary_key=True, nullable=False)

>>> # mapper's perspective of the primary key
>>> b.primary_key
(Column('id', Integer(), table=<user>, primary_key=True, nullable=False),)

>>> # MapperProperties available from .attrs
>>> b.attrs.keys()
['name_syn', 'addresses', 'id', 'name']

>>> # .column_attrs, .relationships, etc. filter this collection
>>> b.column_attrs.keys()
['id', 'name']

>>> list(b.relationships)
[<sqlalchemy.orm.properties.RelationshipProperty object at 0x1015212d0>]

>>> # they are also namespaces
>>> b.column_attrs.id
<sqlalchemy.orm.properties.ColumnProperty object at 0x101525090>

>>> b.relationships.addresses
<sqlalchemy.orm.properties.RelationshipProperty object at 0x1015212d0>

>>> # point inspect() at a mapped, class level attribute,
>>> # returns the attribute itself
>>> b = inspect(User.addresses)
>>> b
<sqlalchemy.orm.attributes.InstrumentedAttribute object at 0x101521fd0>

>>> # From here we can get the mapper:
>>> b.mapper
<Mapper at 0x101525810; Address>

>>> # the parent inspector, in this case a mapper
>>> b.parent
<Mapper at 0x101521950; User>

>>> # an expression
>>> print(b.expression)
"user".id = address.user_id

>>> # inspect works on instances
>>> u1 = User(id=3, name='x')
>>> b = inspect(u1)

>>> # it returns the InstanceState
>>> b
<sqlalchemy.orm.state.InstanceState object at 0x10152bed0>

>>> # similar attrs accessor refers to the
>>> b.attrs.keys()
['id', 'name_syn', 'addresses', 'name']

>>> # attribute interface - from attrs, you get a state object
>>> b.attrs.id
<sqlalchemy.orm.state.AttributeState object at 0x10152bf90>

>>> # this object can give you, current value...
>>> b.attrs.id.value
3

>>> # ... current history
>>> b.attrs.id.history
History(added=[3], unchanged=(), deleted=())

>>> # InstanceState can also provide session state information
>>> # lets assume the object is persistent
>>> s = Session()
>>> s.add(u1)
>>> s.commit()

>>> # now we can get primary key identity, always
>>> # works in query.get()
>>> b.identity
(3,)

>>> # the mapper level key
>>> b.identity_key
(<class '__main__.User'>, (3,))

>>> # state within the session
>>> b.persistent, b.transient, b.deleted, b.detached
(True, False, False, False)

>>> # owning session
>>> b.session
<sqlalchemy.orm.session.Session object at 0x101701150>

from sqlalchemy.orm import with_polymorphic
palias = with_polymorphic(Person, [Engineer, Manager])
session.query(Company).\
            join(palias, Company.employees).\
            filter(or_(Engineer.language=='java', Manager.hair=='pointy'))

# use eager loading in conjunction with with_polymorphic targets
Job_P = with_polymorphic(Job, [SubJob, ExtraJob], aliased=True)
q = s.query(DataContainer).\
            join(DataContainer.jobs.of_type(Job_P)).\
                options(contains_eager(DataContainer.jobs.of_type(Job_P)))

# use eager loading in conjunction with with_polymorphic targets
Job_P = with_polymorphic(Job, [SubJob, ExtraJob], aliased=True)
q = s.query(DataContainer).\
            join(DataContainer.jobs.of_type(Job_P)).\
                options(contains_eager(DataContainer.jobs.of_type(Job_P)))

# pass subclasses to eager loads (implicitly applies with_polymorphic)
q = s.query(ParentThing).\
                options(
                    joinedload_all(
                        ParentThing.container,
                        DataContainer.jobs.of_type(SubJob)
                ))

# control self-referential aliasing with any()/has()
Job_A = aliased(Job)
q = s.query(Job).join(DataContainer.jobs).\
                filter(
                    DataContainer.jobs.of_type(Job_A).\
                        any(and_(Job_A.id < Job.id, Job_A.type=='fred')
                    )
                )

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

@event.listens_for("load", Base, propagate=True)
def on_load(target, context):
    print("New instance loaded:", target)

# on_load() will be applied to SomeClass
class SomeClass(Base):
    __tablename__ = 'sometable'

    # ...

class Snack(Base):
    # ...

    peanuts = relationship("nuts.Peanut",
            primaryjoin="nuts.Peanut.snack_id == Snack.id")

class ReflectedOne(DeferredReflection, Base):
    __abstract__ = True

class ReflectedTwo(DeferredReflection, Base):
    __abstract__ = True

class MyClass(ReflectedOne):
    __tablename__ = 'mytable'

class MyOtherClass(ReflectedOne):
    __tablename__ = 'myothertable'

class YetAnotherClass(ReflectedTwo):
    __tablename__ = 'yetanothertable'

ReflectedOne.prepare(engine_one)
ReflectedTwo.prepare(engine_two)

from sqlalchemy import select

stmt = select([User]).where(User.id == 5)

query(SomeEntity).\
    filter(SomeEntity.id==SomeOtherEntity.id).\
    filter(SomeOtherEntity.foo=='bar').\
    update({"data":"x"})

query(Engineer).\
        filter(Person.id==Engineer.id).\
        filter(Person.name=='dilbert').\
        update({"engineer_data":"java"})

UPDATE engineer SET engineer_data='java' FROM person
WHERE person.id=engineer.id AND person.name='dilbert'

--- examples/beaker_caching/caching_query.py
+++ examples/beaker_caching/caching_query.py
@@ -222,7 +222,8 @@

         """
         if query._current_path:
-            mapper, key = query._current_path[-2:]
+            mapper, prop = query._current_path[-2:]
+            key = prop.key

             for cls in mapper.class_.__mro__:
                 if (cls, key) in self._relationship_options:

from sqlalchemy.types import Numeric
from sqlalchemy.sql import func

class CustomNumeric(Numeric):
    class comparator_factory(Numeric.Comparator):
        def log(self, other):
            return func.log(self.expr, other)

data = Table('data', metadata,
          Column('id', Integer, primary_key=True),
          Column('x', CustomNumeric(10, 5)),
          Column('y', CustomNumeric(10, 5))
     )

stmt = select([data.c.x.log(data.c.y)]).where(data.c.x.log(2) < value)
print(conn.execute(stmt).fetchall())

users.insert().values([
                    {"name": "some name"},
                    {"name": "some other name"},
                    {"name": "yet another name"},
                ])

from sqlalchemy.types import String
from sqlalchemy import func, Table, Column, MetaData

class LowerString(String):
    def bind_expression(self, bindvalue):
        return func.lower(bindvalue)

    def column_expression(self, col):
        return func.lower(col)

metadata = MetaData()
test_table = Table(
        'test_table',
        metadata,
        Column('data', LowerString)
)

>>> print(select([test_table]).where(test_table.c.data == 'HI'))
SELECT lower(test_table.data) AS data
FROM test_table
WHERE test_table.data = lower(:data_1)

from sqlalchemy import inspect
from sqlalchemy import create_engine

engine = create_engine("postgresql://scott:tiger@localhost/test")
insp = inspect(engine)
print(insp.get_table_names())

class SnortEvent(Base):
    __tablename__ = "event"

    id = Column(Integer, primary_key=True)
    signature = Column(Integer, ForeignKey("signature.id"))

    signatures = relationship("Signature", lazy=False)

class Signature(Base):
    __tablename__ = "signature"

    id = Column(Integer, primary_key=True)

    sig_count = column_property(
                    select([func.count('*')]).\
                        where(SnortEvent.signature == id).
                        correlate_except(SnortEvent)
                )

from sqlalchemy.dialects.postgresql import HSTORE

data = Table('data_table', metadata,
        Column('id', Integer, primary_key=True),
        Column('hstore_data', HSTORE)
    )

engine.execute(
    select([data.c.hstore_data['some_key']])
).scalar()

engine.execute(
    select([data.c.hstore_data.matrix()])
).scalar()

# old way, still works since PG supports N-dimensions per row:
Column("my_array", postgresql.ARRAY(Integer))

# new way, will render ARRAY with correct number of [] in DDL,
# will process binds and results more efficiently as we don't need
# to guess how many levels deep to go
Column("my_array", postgresql.ARRAY(Integer, dimensions=2))

result = conn.execute(
    select([mytable.c.arraycol[2]])
)

result = conn.execute(
    select([mytable.c.arraycol[2:4]])
)

conn.execute(
    mytable.update().values({mytable.c.arraycol[2:3]: [7, 8]})
)

>>> from sqlalchemy.dialects import postgresql
>>> conn.scalar(
...    select([
...        postgresql.array([1, 2]) + postgresql.array([3, 4, 5])
...    ])
...  )
[1, 2, 3, 4, 5]

select([mytable.c.arraycol + [4, 5, 6]])

Column('sometimestamp', sqlite.DATETIME(truncate_microseconds=True))
Column('sometimestamp', sqlite.DATETIME(
                    storage_format=(
                                "%(year)04d%(month)02d%(day)02d"
                                "%(hour)02d%(minute)02d%(second)02d%(microsecond)06d"
                    ),
                    regexp="(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})(\d{6})"
                    )
            )
Column('somedate', sqlite.DATE(
                    storage_format="%(month)02d/%(day)02d/%(year)04d",
                    regexp="(?P<month>\d+)/(?P<day>\d+)/(?P<year>\d+)",
                )
            )

>>> stmt = select([cast(sometable.c.somechar, String(20, collation='utf8'))])
>>> print(stmt)
SELECT CAST(sometable.somechar AS VARCHAR(20) COLLATE "utf8") AS anon_1
FROM sometable

stmt = table.delete().prefix_with("LOW_PRIORITY", dialect="mysql")


stmt = table.update().prefix_with("LOW_PRIORITY", dialect="mysql")

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, backref
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

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

class UserKeyword(Base):
    __tablename__ = 'user_keyword'
    user_id = Column(Integer, ForeignKey('user.id'), primary_key=True)
    keyword_id = Column(Integer, ForeignKey('keyword.id'), primary_key=True)

    user = relationship(User,
                backref=backref("user_keywords",
                                cascade="all, delete-orphan")
            )

    keyword = relationship("Keyword",
                backref=backref("user_keywords",
                                cascade="all, delete-orphan")
            )

    # uncomment this to enable the old behavior
    # __mapper_args__ = {"legacy_is_orphan": True}

class Keyword(Base):
    __tablename__ = 'keyword'
    id = Column(Integer, primary_key=True)
    keyword = Column('keyword', String(64))

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

# note we're using PostgreSQL to ensure that referential integrity
# is enforced, for demonstration purposes.
e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)

Base.metadata.drop_all(e)
Base.metadata.create_all(e)

session = Session(e)

u1 = User(name="u1")
k1 = Keyword(keyword="k1")

session.add_all([u1, k1])

uk1 = UserKeyword(keyword=k1, user=u1)

# previously, if session.flush() were called here,
# this operation would succeed, but if session.flush()
# were not called here, the operation fails with an
# integrity error.
# session.flush()
del u1.user_keywords[0]

session.commit()

@event.listens_for(Session, "after_attach")
def after_attach(session, instance):
    assert instance in session

@event.listens_for(Session, "before_attach")
def before_attach(session, instance):
    instance.some_necessary_attribute = session.query(Widget).\
                                            filter_by(instance.widget_name).\
                                            first()

subq = session.query(Entity.value).\
                filter(Entity.id==Parent.entity_id).\
                correlate(Parent).\
                as_scalar()
session.query(Parent).filter(subq=="some value")

subq = session.query(Entity.value).\
                filter(Entity.id==Parent.entity_id).\
                as_scalar()
session.query(Parent).filter(subq=="some value")

from sqlalchemy.sql import table, column, select

t1 = table('t1', column('x'))
t2 = table('t2', column('y'))
s = select([t1, t2]).correlate(t1)

print(s)

SELECT t1.x, t2.y FROM t2

SELECT t1.x, t2.y FROM t1, t2

s2 = select([t1, t2]).where(t1.c.x == t2.c.y).where(t1.c.x == s)

print(s2)

SELECT t1.x, t2.y FROM t1, t2
WHERE t1.x = t2.y AND t1.x =
    (SELECT t1.x, t2.y FROM t2)

scalar_subq = select([someothertable.c.id]).where(someothertable.c.data=='foo')
select([sometable]).where(sometable.c.id==scalar_subq)

s = select([table1]).apply_labels()
s.c.table1_col1
s.c.table1_col2

# before 0.8
table1 = Table('t1', metadata,
    Column('col1', Integer, key='column_one')
)
s = select([table1])
s.c.column_one # would be accessible like this
s.c.col1 # would raise AttributeError

s = select([table1]).apply_labels()
s.c.table1_column_one # would raise AttributeError
s.c.table1_col1 # would be accessible like this

# with 0.8
table1 = Table('t1', metadata,
    Column('col1', Integer, key='column_one')
)
s = select([table1])
s.c.column_one # works
s.c.col1 # AttributeError

s = select([table1]).apply_labels()
s.c.table1_column_one # works
s.c.table1_col1 # AttributeError

@event.listens_for(Table, "column_reflect")
def listen_for_col(inspector, table, column_info):
    # ...

t1 = table('t1', column('x'))
t1.insert().values(x=5, z=5) # raises "Unconsumed column names: z"

>>> insp.get_primary_keys()
["a", "b"]

>>> insp.get_pk_constraint()
{"name":"pk_constraint", "constrained_columns":["a", "b"]}

>>> row = result.fetchone()
>>> row['foo'] == row['FOO'] == row['Foo']
True