SqlAlchemy 2.0 中文文档（七十六）（1）

from sqlalchemy.ext import baked
from sqlalchemy import bindparam
bakery = baked.bakery()
def search_for_user(session, username, email=None):
    baked_query = bakery(lambda session: session.query(User))
    baked_query += lambda q: q.filter(User.name == bindparam("username"))
    baked_query += lambda q: q.order_by(User.id)
    if email:
        baked_query += lambda q: q.filter(User.email == bindparam("email"))
    result = baked_query(session).params(username=username, email=email).all()
    return result

class HasFooBar(object):
    foobar = Column(Integer)
    @declared_attr
    def foobar_prop(cls):
        return column_property("foobar: " + cls.foobar)
class SomeClass(HasFooBar, Base):
    __tablename__ = "some_table"
    id = Column(Integer, primary_key=True)

class HasFooBar(object):
    @declared_attr
    def foobar(cls):
        return Column(Integer)
    @declared_attr
    def foobar_prop(cls):
        return column_property("foobar: " + cls.foobar)
class SomeClass(HasFooBar, Base):
    __tablename__ = "some_table"
    id = Column(Integer, primary_key=True)

class HasIdMixin(object):
    @declared_attr.cascading
    def id(cls):
        if has_inherited_table(cls):
            return Column(ForeignKey("myclass.id"), primary_key=True)
        else:
            return Column(Integer, primary_key=True)
class MyClass(HasIdMixin, Base):
    __tablename__ = "myclass"
    # ...
class MySubClass(MyClass):
  """ """
    # ...

from sqlalchemy import Column, Integer, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import (
    declarative_base,
    declared_attr,
    AbstractConcreteBase,
)
Base = declarative_base()
class Something(Base):
    __tablename__ = "something"
    id = Column(Integer, primary_key=True)
class Abstract(AbstractConcreteBase, Base):
    id = Column(Integer, primary_key=True)
    @declared_attr
    def something_id(cls):
        return Column(ForeignKey(Something.id))
    @declared_attr
    def something(cls):
        return relationship(Something)
class Concrete(Abstract):
    __tablename__ = "cca"
    __mapper_args__ = {"polymorphic_identity": "cca", "concrete": True}

import time
from sqlalchemy import Integer, Column, create_engine, Table
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Foo(Base):
    __table__ = Table(
        "foo",
        Base.metadata,
        Column("id", Integer, primary_key=True),
        Column("a", Integer(), nullable=False),
        Column("b", Integer(), nullable=False),
        Column("c", Integer(), nullable=False),
    )
engine = create_engine("mysql+mysqldb://scott:tiger@localhost/test", echo=True)
sess = Session(engine)
now = time.time()
# avoid using all() so that we don't have the overhead of building
# a large list of full objects in memory
for obj in sess.query(Foo).yield_per(100).limit(1000000):
    pass
print("Total time: %d" % (time.time() - now))

rows = sess.query(Foo.a, Foo.b, Foo.c).all()

-----------------
size=10 num=10000                 # few rows, lots of queries
namedtuple: 3.60302400589         # namedtuple falls over
keyedtuple: 0.255059957504        # KeyedTuple very fast
lw keyed tuple: 0.582715034485    # lw keyed trails right on KeyedTuple
-----------------
size=100 num=1000                 # <--- sweet spot
namedtuple: 0.365247011185
keyedtuple: 0.24896979332
lw keyed tuple: 0.0889317989349   # lw keyed blows both away!
-----------------
size=10000 num=100
namedtuple: 0.572599887848
keyedtuple: 2.54251694679
lw keyed tuple: 0.613876104355
-----------------
size=1000000 num=10               # few queries, lots of rows
namedtuple: 5.79669594765         # namedtuple very fast
keyedtuple: 28.856498003          # KeyedTuple falls over
lw keyed tuple: 6.74346804619     # lw keyed trails right on namedtuple

# reported by heapy, summation of SQLAlchemy objects +
# associated dicts + weakref-related objects with core of Nova imported:
    Before: total count 26477 total bytes 7975712
    After: total count 18181 total bytes 4236456
# reported for the Python module space overall with the
# core of Nova imported:
    Before: Partition of a set of 355558 objects. Total size = 61661760 bytes.
    After: Partition of a set of 346034 objects. Total size = 57808016 bytes.
```### UPDATE 语句现在在刷新中与 executemany() 批处理
现在可以将 UPDATE 语句批处理到 ORM 刷新中，以更高效的 executemany() 调用执行，类似于 INSERT 语句可以批处理；这将根据以下标准在刷新中调用：
+   两个或更多连续的 UPDATE 语句涉及相同的要修改的列集。
+   语句在 SET 子句中没有嵌入的 SQL 表达式。
+   映射不使用 `mapper.version_id_col`，或者后端方言支持 executemany() 操作的“合理”行数；大多数 DBAPI 现在正确支持这一点。### Session.get_bind() 处理更广泛的继承场景
每当查询或工作单元刷新过程寻找与特定类对应的数据库引擎时，都会调用 `Session.get_bind()` 方法。该方法已经改进，以处理各种继承导向的场景，包括：
+   绑定到一个 Mixin 或抽象类：
    ```py
    class MyClass(SomeMixin, Base):
        __tablename__ = "my_table"
        # ...
    session = Session(binds={SomeMixin: some_engine})
    ```
+   基于表格，分别绑定到继承的具体子类：
    ```py
    class BaseClass(Base):
        __tablename__ = "base"
        # ...
    class ConcreteSubClass(BaseClass):
        __tablename__ = "concrete"
        # ...
        __mapper_args__ = {"concrete": True}
    session = Session(binds={base_table: some_engine, concrete_table: some_other_engine})
    ```
[#3035](https://www.sqlalchemy.org/trac/ticket/3035)  ### 在所有相关的查询情况下，Session.get_bind()将接收到 Mapper
修复了一系列问题，其中`Session.get_bind()`未接收到`Query`的主要`Mapper`，尽管此映射器是 readily available 的（主映射器是与`Query`对象关联的单个映射器，或者替代是与查询关联的第一个映射器）。
当传递给`Session.get_bind()`的`Mapper`对象通常由使用`Session.binds`参数的会话使用，以将映射器与一系列引擎关联（虽然在这种用例中，通常情况下“工作”，因为绑定通常会通过映射的表对象找到），或者更具体地实现一个用户定义的`Session.get_bind()`方法，该方法基于映射器提供一些选择引擎的模式，例如水平分片或所谓的“路由”会话，将查询路由到不同的后端。
这些场景包括：
+   `Query.count()`:
    ```py
    session.query(User).count()
    ```
+   `Query.update()` 和 `Query.delete()`，用于 UPDATE/DELETE 语句以及“fetch”策略所使用的 SELECT：
    ```py
    session.query(User).filter(User.id == 15).update(
        {"name": "foob"}, synchronize_session="fetch"
    )
    session.query(User).filter(User.id == 15).delete(synchronize_session="fetch")
    ```
+   对个别列的查询：
    ```py
    session.query(User.id, User.name).all()
    ```
+   对间接映射（例如`column_property`）的 SQL 函数和其他表达式：
    ```py
    class User(Base):
        ...
        score = column_property(func.coalesce(self.tables.users.c.name, None))
    session.query(func.max(User.score)).scalar()
    ```
[#3227](https://www.sqlalchemy.org/trac/ticket/3227) [#3242](https://www.sqlalchemy.org/trac/ticket/3242) [#1326](https://www.sqlalchemy.org/trac/ticket/1326)  ### .info 字典改进
`InspectionAttr.info` 集合现在可用于从`Mapper.all_orm_descriptors`集合中检索到的每种对象。这包括`hybrid_property`和`association_proxy()`。然而，由于这些对象是类绑定的描述符，必须**分开**从它们附加到的类中访问以获取属性。以下是使用`Mapper.all_orm_descriptors`命名空间进行说明：
```py
class SomeObject(Base):
    # ...
    @hybrid_property
    def some_prop(self):
        return self.value + 5
inspect(SomeObject).all_orm_descriptors.some_prop.info["foo"] = "bar"

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
A.b = column_property(select([func.max(B.id)]).where(B.a_id == A.id).correlate(A))

print(sess.query(A, a1).order_by(a1.b))

SELECT  a.id  AS  a_id,  (SELECT  max(b.id)  AS  max_1  FROM  b
WHERE  b.a_id  =  a.id)  AS  anon_1,  a_1.id  AS  a_1_id,
(SELECT  max(b.id)  AS  max_2
FROM  b  WHERE  b.a_id  =  a_1.id)  AS  anon_2
FROM  a,  a  AS  a_1  ORDER  BY  anon_1

SELECT  a.id  AS  a_id,  (SELECT  max(b.id)  AS  max_1
FROM  b  WHERE  b.a_id  =  a.id)  AS  anon_1,  a_1.id  AS  a_1_id,
(SELECT  max(b.id)  AS  max_2
FROM  b  WHERE  b.a_id  =  a_1.id)  AS  anon_2
FROM  a,  a  AS  a_1  ORDER  BY  anon_2

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    type = Column(String)
    __mapper_args__ = {"polymorphic_on": type, "with_polymorphic": "*"}

SELECT  a.id  AS  a_id,  a.type  AS  a_type,  (SELECT  max(b.id)  AS  max_1
FROM  b  WHERE  b.a_id  =  a.id)  AS  anon_1
FROM  a  ORDER  BY  (SELECT  max(b.id)  AS  max_2
FROM  b  WHERE  b.a_id  =  a.id)

SELECT  a.id  AS  a_id,  a.type  AS  a_type,  (SELECT  max(b.id)  AS  max_1
FROM  b  WHERE  b.a_id  =  a.id)  AS  anon_1
FROM  a  ORDER  BY  anon_1

sel = select([table]).limit(bindparam("mylimit")).offset(bindparam("myoffset"))

>>> result = connection.execute(stmt)
>>> result.fetchone()
(1, 'x')
>>> result.fetchone()
None  # indicates no more rows
>>> result.fetchone()
exception: ResourceClosedError

>>> result = connection.execute(stmt)
>>> result.fetchone()
(1, 'x')
>>> result.fetchone()
None  # indicates no more rows
>>> result.fetchone()
None  # still None
>>> result.fetchall()
[]
>>> result.close()
>>> result.fetchone()
exception: ResourceClosedError  # *now* it raises

metadata = MetaData(naming_convention={"ck": "ck_%(table_name)s_%(column_0_name)s"})
foo = Table("foo", metadata, Column("value", Integer))
CheckConstraint(foo.c.value > 5)

CREATE  TABLE  foo  (
  value  INTEGER,
  CONSTRAINT  ck_foo_value  CHECK  (value  >  5)
)

from sqlalchemy import Table, Column, MetaData, Integer, UniqueConstraint
m = MetaData()
t = Table("t", m, Column("a", Integer), Column("b", Integer))
uq = UniqueConstraint(t.c.a, t.c.b)  # will auto-attach to Table
assert uq in t.constraints

from sqlalchemy import Table, Column, MetaData, Integer, UniqueConstraint
m = MetaData()
a = Column("a", Integer)
b = Column("b", Integer)
uq = UniqueConstraint(a, b)
t = Table("t", m, a, b)
assert uq in t.constraints  # constraint auto-attached

from sqlalchemy import Table, Column, MetaData, Integer, UniqueConstraint
m = MetaData()
a = Column("a", Integer)
b = Column("b", Integer)
uq = UniqueConstraint(a, "b")
t = Table("t", m, a, b)
# constraint *not* auto-attached, as we do not have tracking
# to locate when a name 'b' becomes available on the table
assert uq not in t.constraints

from sqlalchemy import Table, Column, MetaData, Integer, UniqueConstraint
m = MetaData()
a = Column("a", Integer)
b = Column("b", Integer)
t = Table("t", m, a, b)
uq = UniqueConstraint(a, "b")
# constraint auto-attached normally as in older versions
assert uq in t.constraints

from sqlalchemy import Table, Column, MetaData, Integer, select, func
m = MetaData()
t = Table(
    "t", m, Column("x", Integer), Column("y", Integer, default=func.somefunction())
)
stmt = select([t.c.x])
print(t.insert().from_select(["x"], stmt))

INSERT  INTO  t  (x,  y)  SELECT  t.x,  somefunction()  AS  somefunction_1
FROM  t

from sqlalchemy import Table, Column, MetaData, Text
from sqlalchemy.schema import CreateTable
from sqlalchemy.dialects.postgresql import ARRAY, array
from sqlalchemy.dialects import postgresql
metadata = MetaData()
tbl = Table(
    "derp",
    metadata,
    Column("arr", ARRAY(Text), server_default=array(["foo", "bar", "baz"])),
)
print(CreateTable(tbl).compile(dialect=postgresql.dialect()))

CREATE  TABLE  derp  (
  arr  TEXT[]  DEFAULT  ARRAY['foo',  'bar',  'baz']
)

from sqlalchemy import create_engine, Unicode, select, cast
import random
import warnings
e = create_engine("sqlite://")
# Use the "once" filter (which is also the default for Python
# warnings).  Exactly ten of these warnings will
# be emitted; beyond that, the Python warnings registry will accumulate
# new values as dupes of one of the ten existing.
warnings.filterwarnings("once")
for i in range(1000):
    e.execute(
        select([cast(("foo_%d" % random.randint(0, 1000000)).encode("ascii"), Unicode)])
    )

/path/lib/sqlalchemy/sql/sqltypes.py:186: SAWarning: Unicode type received
  non-unicode bind param value 'foo_4852'. (this warning may be
  suppressed after 10 occurrences)