SqlAlchemy 2.0 中文文档（四十四）（2）

with engine.connect() as conn:
    with conn.execution_options(stream_results=True, max_row_buffer=100).execute(
        text("select * from table")
    ) as result:
        for row in result:
            print(f"{row}")

user_table = Table(
    "user",
    metadata_obj,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
)

connection = engine.connect().execution_options(
    schema_translate_map={None: "user_schema_one"}
)
result = connection.execute(user_table.select())

SELECT  user_schema_one.user.id,  user_schema_one.user.name  FROM
user_schema_one.user

connection = engine.connect().execution_options(
    schema_translate_map={
        None: "user_schema_one",  # no schema name -> "user_schema_one"
        "special": "special_schema",  # schema="special" becomes "special_schema"
        "public": None,  # Table objects with schema="public" will render with no schema
    }
)

schema_engine = engine.execution_options(schema_translate_map={...})
session = Session(schema_engine)
...

def run_my_statement(connection, parameter):
    stmt = select(table)
    stmt = stmt.where(table.c.col == parameter)
    stmt = stmt.order_by(table.c.id)
    return connection.execute(stmt)

engine = create_engine(
    "postgresql+psycopg2://scott:tiger@localhost/test", query_cache_size=1200
)

from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
Base = declarative_base()
class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    data = Column(String)
    bs = relationship("B")
class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    data = Column(String)
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
s = Session(e)
s.add_all([A(bs=[B(), B(), B()]), A(bs=[B(), B(), B()]), A(bs=[B(), B(), B()])])
s.commit()
for a_rec in s.scalars(select(A)):
    print(a_rec.bs)

INFO sqlalchemy.engine.Engine PRAGMA temp.table_info("a")
INFO sqlalchemy.engine.Engine [raw sql] ()
INFO sqlalchemy.engine.Engine PRAGMA main.table_info("b")
INFO sqlalchemy.engine.Engine [raw sql] ()

INFO  sqlalchemy.engine.Engine
CREATE  TABLE  a  (
  id  INTEGER  NOT  NULL,
  data  VARCHAR,
  PRIMARY  KEY  (id)
)
INFO  sqlalchemy.engine.Engine  [no  key  0.00007s]  ()
INFO  sqlalchemy.engine.Engine
CREATE  TABLE  b  (
  id  INTEGER  NOT  NULL,
  a_id  INTEGER,
  data  VARCHAR,
  PRIMARY  KEY  (id),
  FOREIGN  KEY(a_id)  REFERENCES  a  (id)
)
INFO  sqlalchemy.engine.Engine  [no  key  0.00006s]  ()

INFO  sqlalchemy.engine.Engine  INSERT  INTO  a  (data)  VALUES  (?)
INFO  sqlalchemy.engine.Engine  [generated  in  0.00011s]  (None,)
INFO  sqlalchemy.engine.Engine  INSERT  INTO  a  (data)  VALUES  (?)
INFO  sqlalchemy.engine.Engine  [cached  since  0.0003533s  ago]  (None,)
INFO  sqlalchemy.engine.Engine  INSERT  INTO  a  (data)  VALUES  (?)
INFO  sqlalchemy.engine.Engine  [cached  since  0.0005326s  ago]  (None,)
INFO  sqlalchemy.engine.Engine  INSERT  INTO  b  (a_id,  data)  VALUES  (?,  ?)
INFO  sqlalchemy.engine.Engine  [generated  in  0.00010s]  (1,  None)
INFO  sqlalchemy.engine.Engine  INSERT  INTO  b  (a_id,  data)  VALUES  (?,  ?)
INFO  sqlalchemy.engine.Engine  [cached  since  0.0003232s  ago]  (1,  None)
INFO  sqlalchemy.engine.Engine  INSERT  INTO  b  (a_id,  data)  VALUES  (?,  ?)
INFO  sqlalchemy.engine.Engine  [cached  since  0.0004887s  ago]  (1,  None)

INFO sqlalchemy.engine.Engine SELECT a.id AS a_id, a.data AS a_data
FROM a
INFO sqlalchemy.engine.Engine [generated in 0.00009s] ()
INFO sqlalchemy.engine.Engine SELECT b.id AS b_id, b.a_id AS b_a_id, b.data AS b_data
FROM b
WHERE ? = b.a_id
INFO sqlalchemy.engine.Engine [generated in 0.00010s] (1,)
INFO sqlalchemy.engine.Engine SELECT b.id AS b_id, b.a_id AS b_a_id, b.data AS b_data
FROM b
WHERE ? = b.a_id
INFO sqlalchemy.engine.Engine [cached since 0.0005922s ago] (2,)
INFO sqlalchemy.engine.Engine SELECT b.id AS b_id, b.a_id AS b_a_id, b.data AS b_data
FROM b
WHERE ? = b.a_id

my_cache = {}
with engine.connect().execution_options(compiled_cache=my_cache) as conn:
    conn.execute(table.select())

# disable caching for this connection
with engine.connect().execution_options(compiled_cache=None) as conn:
    conn.execute(table.select())
```### 第三方方言的缓存
缓存功能要求方言的编译器生成的 SQL 字符串可以安全地用于许多语句调用，给定一个特定的缓存键，该键与该 SQL 字符串相关联。这意味着语句中的任何字面值，例如 SELECT 的 LIMIT/OFFSET 值，不能在方言的编译方案中硬编码，因为编译后的字符串将无法重用。SQLAlchemy 支持使用`BindParameter.render_literal_execute()`方法渲染绑定参数，该方法可以应用于现有的 `Select._limit_clause` 和 `Select._offset_clause` 属性，由自定义编译器实现，后文将对此进行说明。
由于有许多第三方方言，其中许多可能会从 SQL 语句中生成字面值，而不使用更新的“字面执行”功能，因此自 SQLAlchemy 1.4.5 起，为方言添加了一个名为`Dialect.supports_statement_cache`的属性。此属性在运行时直接在特定方言类上检查其是否存在，即使它已经存在于超类上，因此，即使第三方方言是现有可缓存的 SQLAlchemy 方言的子类，例如`sqlalchemy.dialects.postgresql.PGDialect`，仍必须显式包含此属性以启用缓存。该属性应仅在方言经过必要的修改并测试过使用不同参数的编译 SQL 语句的可重用性后才能**启用**。
对于所有不支持此属性的第三方方言，其日志将指示`方言不支持缓存`。
当一个方言已经针对缓存进行了测试，特别是 SQL 编译器已更新为不直接在 SQL 字符串中渲染任何字面值 LIMIT / OFFSET 时，方言作者可以如下应用该属性：
```py
from sqlalchemy.engine.default import DefaultDialect
class MyDialect(DefaultDialect):
    supports_statement_cache = True

class MyDBAPIForMyDialect(MyDialect):
    supports_statement_cache = True

# pre 1.4 style code
def limit_clause(self, select, **kw):
    text = ""
    if select._limit is not None:
        text += " \n LIMIT %d" % (select._limit,)
    if select._offset is not None:
        text += " \n OFFSET %d" % (select._offset,)
    return text

# 1.4 cache-compatible code
def limit_clause(self, select, **kw):
    text = ""
    limit_clause = select._limit_clause
    offset_clause = select._offset_clause
    if select._simple_int_clause(limit_clause):
        text += " \n LIMIT %s" % (
            self.process(limit_clause.render_literal_execute(), **kw)
        )
    elif limit_clause is not None:
        # assuming the DB doesn't support SQL expressions for LIMIT.
        # Otherwise render here normally
        raise exc.CompileError(
            "dialect 'mydialect' can only render simple integers for LIMIT"
        )
    if select._simple_int_clause(offset_clause):
        text += " \n OFFSET %s" % (
            self.process(offset_clause.render_literal_execute(), **kw)
        )
    elif offset_clause is not None:
        # assuming the DB doesn't support SQL expressions for OFFSET.
        # Otherwise render here normally
        raise exc.CompileError(
            "dialect 'mydialect' can only render simple integers for OFFSET"
        )
    return text

SELECT  x  FROM  y
LIMIT  __[POSTCOMPILE_param_1]
OFFSET  __[POSTCOMPILE_param_2]

from sqlalchemy import lambda_stmt
def run_my_statement(connection, parameter):
    stmt = lambda_stmt(lambda: select(table))
    stmt += lambda s: s.where(table.c.col == parameter)
    stmt += lambda s: s.order_by(table.c.id)
    return connection.execute(stmt)
with engine.connect() as conn:
    result = run_my_statement(some_connection, "some parameter")

def upd(id_, newname):
    stmt = lambda_stmt(lambda: users.update())
    stmt += lambda s: s.values(name=newname)
    stmt += lambda s: s.where(users.c.id == id_)
    return stmt
with engine.begin() as conn:
    conn.execute(upd(7, "foo"))

def select_user(session, name):
    stmt = lambda_stmt(lambda: select(User))
    stmt += lambda s: s.where(User.name == name)
    row = session.execute(stmt).first()
    return row

>>> def my_stmt(x, y):
...     stmt = lambda_stmt(lambda: select(func.max(x, y)))
...     return stmt
>>> engine = create_engine("sqlite://", echo=True)
>>> with engine.connect() as conn:
...     print(conn.scalar(my_stmt(5, 10)))
...     print(conn.scalar(my_stmt(12, 8)))
SELECT  max(?,  ?)  AS  max_1
[generated  in  0.00057s]  (5,  10)
10
SELECT  max(?,  ?)  AS  max_1
[cached  since  0.002059s  ago]  (12,  8)
12

# **Don't** do this:
def my_stmt(parameter, thing=False):
    stmt = lambda_stmt(lambda: select(table))
    stmt += lambda s: (
        s.where(table.c.x > parameter) if thing else s.where(table.c.y == parameter)
    )
    return stmt
# **Do** do this:
def my_stmt(parameter, thing=False):
    stmt = lambda_stmt(lambda: select(table))
    if thing:
        stmt += lambda s: s.where(table.c.x > parameter)
    else:
        stmt += lambda s: s.where(table.c.y == parameter)
    return stmt

>>> def my_stmt(x, y):
...     def get_x():
...         return x
...
...     def get_y():
...         return y
...
...     stmt = lambda_stmt(lambda: select(func.max(get_x(), get_y())))
...     return stmt
>>> with engine.connect() as conn:
...     print(conn.scalar(my_stmt(5, 10)))
Traceback (most recent call last):
 # ...
sqlalchemy.exc.InvalidRequestError: Can't invoke Python callable get_x()
inside of lambda expression argument at
<code object <lambda> at 0x7fed15f350e0, file "<stdin>", line 6>;
lambda SQL constructs should not invoke functions from closure variables
to produce literal values since the lambda SQL system normally extracts
bound values without actually invoking the lambda or any functions within it.

>>> def my_stmt(x, y):
...     def get_x():
...         return x
...
...     def get_y():
...         return y
...
...     x_param, y_param = get_x(), get_y()
...     stmt = lambda_stmt(lambda: select(func.max(x_param, y_param)))
...     return stmt

>>> class Foo:
...     def __init__(self, x, y):
...         self.x = x
...         self.y = y
>>> def my_stmt(foo):
...     stmt = lambda_stmt(lambda: select(func.max(foo.x, foo.y)))
...     return stmt
>>> with engine.connect() as conn:
...     print(conn.scalar(my_stmt(Foo(5, 10))))
Traceback (most recent call last):
 # ...
sqlalchemy.exc.InvalidRequestError: Closure variable named 'foo' inside of
lambda callable <code object <lambda> at 0x7fed15f35450, file
"<stdin>", line 2> does not refer to a cacheable SQL element, and also
does not appear to be serving as a SQL literal bound value based on the
default SQL expression returned by the function.  This variable needs to
remain outside the scope of a SQL-generating lambda so that a proper cache
key may be generated from the lambda's state.  Evaluate this variable
outside of the lambda, set track_on=[<elements>] to explicitly select
closure elements to track, or set track_closure_variables=False to exclude
closure variables from being part of the cache key.

>>> def my_stmt(foo):
...     x_param, y_param = foo.x, foo.y
...     stmt = lambda_stmt(lambda: select(func.max(x_param, y_param)))
...     return stmt

>>> def my_stmt(foo):
...     stmt = lambda_stmt(
...         lambda: select(func.max(foo.x, foo.y)), track_closure_variables=False
...     )
...     return stmt

>>> def my_stmt(self, foo):
...     stmt = lambda_stmt(
...         lambda: select(*self.column_expressions), track_closure_variables=False
...     )
...     stmt = stmt.add_criteria(lambda: self.where_criteria, track_on=[self])
...     return stmt

>>> from sqlalchemy import select, column
>>> stmt = select(column("q"))
>>> cache_key = stmt._generate_cache_key()
>>> print(cache_key)  # somewhat paraphrased
CacheKey(key=(
 '0',
 <class 'sqlalchemy.sql.selectable.Select'>,
 '_raw_columns',
 (
 (
 '1',
 <class 'sqlalchemy.sql.elements.ColumnClause'>,
 'name',
 'q',
 'type',
 (
 <class 'sqlalchemy.sql.sqltypes.NullType'>,
 ),
 ),
 ),
 # a few more elements are here, and many more for a more
 # complicated SELECT statement
),)

>>> from sqlalchemy import lambda_stmt
>>> stmt = lambda_stmt(lambda: select(column("q")))
>>> cache_key = stmt._generate_cache_key()
>>> print(cache_key)
CacheKey(key=(
 <code object <lambda> at 0x7fed1617c710, file "<stdin>", line 1>,
 <class 'sqlalchemy.sql.lambdas.StatementLambdaElement'>,
),)

>>> def my_stmt(parameter):
...     col = column("q")
...     stmt = lambda_stmt(lambda: select(col))
...     stmt += lambda s: s.where(col == parameter)
...     return stmt

>>> stmt = my_stmt(5)
>>> key = stmt._generate_cache_key()
>>> print(key)
CacheKey(key=(
 <code object <lambda> at 0x7f07323c50e0, file "<stdin>", line 3>,
 (
 '0',
 <class 'sqlalchemy.sql.elements.ColumnClause'>,
 'name',
 'q',
 'type',
 (
 <class 'sqlalchemy.sql.sqltypes.NullType'>,
 ),
 ),
 <code object <lambda> at 0x7f07323c5190, file "<stdin>", line 4>,
 <class 'sqlalchemy.sql.lambdas.LinkedLambdaElement'>,
 (
 '0',
 <class 'sqlalchemy.sql.elements.ColumnClause'>,
 'name',
 'q',
 'type',
 (
 <class 'sqlalchemy.sql.sqltypes.NullType'>,
 ),
 ),
 (
 '0',
 <class 'sqlalchemy.sql.elements.ColumnClause'>,
 'name',
 'q',
 'type',
 (
 <class 'sqlalchemy.sql.sqltypes.NullType'>,
 ),
 ),
),)

>>> key.bindparams
[BindParameter('%(139668884281280 parameter)s', 5, type_=Integer())]

engine = create_engine(
    "mariadb+mariadbconnector://scott:tiger@host/db", use_insertmanyvalues=False
)

t = Table(
    "t",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("x", Integer),
    implicit_returning=False,
)

INSERT  INTO  a  (data,  x,  y)  VALUES  (%(data)s,  %(x)s,  %(y)s)  RETURNING  a.id

INSERT  INTO  a  (data,  x,  y)  VALUES
  (%(data_0)s,  %(x_0)s,  %(y_0)s),
  (%(data_1)s,  %(x_1)s,  %(y_1)s),
  (%(data_2)s,  %(x_2)s,  %(y_2)s),
  ...
  (%(data_78)s,  %(x_78)s,  %(y_78)s)
RETURNING  a.id