基本用法¶

回忆起 引擎配置 那是 Engine 通过创建 create_engine() 呼叫：

engine = create_engine('mysql://scott:tiger@localhost/test')

的典型用法 create_engine() 是每个特定数据库URL一次，在单个应用程序进程的生命周期内全局保存。单一的 Engine 管理许多个人 DBAPI 代表进程的连接，并打算以并发方式调用。这个 Engine 是 not 与DBAPI同义 connect 函数，它只表示一个连接资源 Engine 仅在应用程序的模块级创建一次，而不是每个对象或每个函数调用时，效率最高。

最基本的功能 Engine 提供访问 Connection ，然后可以调用SQL语句。向数据库发出文本语句如下所示：

from sqlalchemy import text

with engine.connect() as connection:
    result = connection.execute(text("select username from users"))
    for row in result:
        print("username:", row['username'])

上面， Engine.connect() 方法返回 Connection 对象，并在Python上下文管理器中使用它（例如 with: 声明）the Connection.close() 方法在块末尾自动调用。这个 Connection 是一个 代理 对象获取实际的DBAPI连接。DBAPI连接是在 Connection 创建。

返回的对象称为 CursorResult ，它引用DBAPI游标，并提供用于获取与DBAPI游标类似的行的方法。DBAPI光标将由 CursorResult 当它的所有结果行（如果有）用完时。一 CursorResult 它不返回任何行，例如UPDATE语句的行（不返回任何行），在构造后立即释放光标资源。

当 Connection 在结束时关闭 with: 块，引用的DBAPI连接是 released 到连接池。从数据库本身的角度来看，假设连接池有空间存储此连接以备下次使用，那么连接池实际上不会“关闭”连接。当连接返回池以供重新使用时，池机制将发出一个 rollback() 调用DBAPI连接，以便删除所有事务状态或锁，并且该连接已准备好供下次使用。

上面的示例演示了文本SQL字符串的执行，应该使用 text() 构造来指示我们希望使用文本SQL。这个 Connection.execute() 方法当然可以容纳更多的内容，包括中描述的各种SQL表达式构造。 SQL表达式语言教程（1.x API） .

使用事务¶

这个 Connection 对象提供 Connection.begin() 返回 Transaction 对象。就像 Connection 它本身通常在Python中使用 with: 块以便管理其作用域：

with engine.connect() as connection:
    with connection.begin():
        r1 = connection.execute(table1.select())
        connection.execute(table1.insert(), {"col1": 7, "col2": "this is some data"})

上面的块可以通过使用 Engine.begin() 方法 Engine ：：

# runs a transaction
with engine.begin() as connection:
    r1 = connection.execute(table1.select())
    connection.execute(table1.insert(), {"col1": 7, "col2": "this is some data"})

各自管理的区块 .begin() 方法的行为使得在块完成时提交事务。如果引发异常，则会回滚事务，并向外传播异常。

用于表示事务的基础对象是 Transaction 对象。此对象由 Connection.begin() 方法，包括方法 Transaction.commit() 和 Transaction.rollback() . 推荐使用上下文管理器调用表单（它自动调用这些方法）作为最佳实践。

# method_a starts a transaction and calls method_b
def method_a(connection):
    with connection.begin():  # open a transaction
        method_b(connection)

# method_b also starts a transaction
def method_b(connection):
    with connection.begin(): # open a transaction - this runs in the
                             # context of method_a's transaction
        connection.execute(text("insert into mytable values ('bat', 'lala')"))
        connection.execute(mytable.insert(), {"col1": "bat", "col2": "lala"})

# open a Connection and call method_a
with engine.connect() as conn:
    method_a(conn)

# method_a calls method_b
def method_a(connection):
    method_b(connection)

# method_b uses the connection and assumes the transaction
# is external
def method_b(connection):
    connection.execute(text("insert into mytable values ('bat', 'lala')"))
    connection.execute(mytable.insert(), {"col1": "bat", "col2": "lala"})

# open a Connection inside of a transaction and call method_a
with engine.begin() as conn:
    method_a(conn)

import contextlib

@contextlib.contextmanager
def transaction(connection):
    if not connection.in_transaction():
        with connection.begin():
            yield connection
    else:
        yield connection

# method_a starts a transaction and calls method_b
def method_a(connection):
    with transaction(connection):  # open a transaction
        method_b(connection)

# method_b either starts a transaction, or uses the one already
# present
def method_b(connection):
    with transaction(connection):  # open a transaction
        connection.execute(text("insert into mytable values ('bat', 'lala')"))
        connection.execute(mytable.insert(), {"col1": "bat", "col2": "lala"})

# open a Connection and call method_a
with engine.connect() as conn:
    method_a(conn)

import contextlib

def connectivity(engine):
    connection = None

    @contextlib.contextmanager
    def connect():
        nonlocal connection

        if connection is None:
            connection = engine.connect()
            with connection:
                with connection.begin():
                    yield connection
        else:
            yield connection

    return connect

# method_a passes along connectivity context, at the same time
# it chooses to establish a connection by calling "with"
def method_a(connectivity):
    with connectivity():
        method_b(connectivity)

# method_b also wants to use a connection from the context, so it
# also calls "with:", but also it actually uses the connection.
def method_b(connectivity):
    with connectivity() as connection:
        connection.execute(text("insert into mytable values ('bat', 'lala')"))
        connection.execute(mytable.insert(), {"col1": "bat", "col2": "lala"})

# create a new connection/transaction context object and call
# method_a
method_a(connectivity(engine))

库级（例如仿真）自动提交¶

上一个事务示例说明了如何使用 Transaction 以便多个执行可以参与同一事务。如果不使用 Transaction ？虽然一些DBAPI实现提供了各种特殊的“非事务”模式，但根据PEP-0249，DBAPI的核心行为是 事务始终在进行中 ，仅提供 rollback() 和 commit() 方法但不 begin() . SQLAlchemy假定任何给定DBAPI都是这样。

考虑到这一需求，SQLAlchemy实现了自己的“自动提交”功能，该功能可以在所有后端完全一致地工作。这是通过检测表示数据更改操作的语句（即插入、更新、删除）以及数据定义语言（DDL）语句（如创建表、更改表）来实现的，然后在没有事务进行时自动发出提交。检测是基于 autocommit=True 语句的执行选项。如果该语句是纯文本语句且未设置标志，则使用正则表达式检测特定后端的插入、更新、删除以及各种其他命令：

conn = engine.connect()
conn.execute(text("INSERT INTO users VALUES (1, 'john')"))  # autocommits

“自动提交”功能仅在“否”时有效 Transaction 已另行声明。这意味着该特性通常不与ORM一起使用，因为 Session 默认情况下，对象始终保持 Transaction .

使用Generative可以完全控制“自动提交”行为 Connection.execution_options() 方法提供于 Connection 和 Engine ，使用“autocommit”标志打开或关闭所选作用域的自动提交。例如，A text() 表示提交的存储过程的构造可能会使用它，以便select语句将发出commit:：

with engine.connect().execution_options(autocommit=True) as conn:
    conn.execute(text("SELECT my_mutating_procedure()"))

设置事务隔离级别，包括DBAPI Autocommit¶

大多数dbapi支持可配置事务的概念 isolation 水平。这通常是四个级别“未提交读”、“已提交读”、“可重复读”和“可序列化”。它们通常在DBAPI连接开始一个新事务之前应用于它，注意大多数DBAPI将在SQL语句首次发出时隐式地开始该事务。

支持隔离级别的DBAPI通常还支持真正的“自动提交”概念，这意味着DBAPI连接本身将被置于非事务性自动提交模式。这通常意味着自动向数据库发出“BEGIN”的典型DBAPI行为不再发生，但也可能包括其他指令。SQLAlChemy对待“自动提交”的概念与对待任何其他隔离级别一样；因为它是一个不仅失去“读提交”而且失去原子性的隔离级别。

SQLAlchemy方言应该支持这些隔离级别，并尽可能地自动提交。这些级别是通过一系列贯穿核心的“执行选项”参数和方法来设置的，例如 Connection.execution_options() 方法。该参数称为 Connection.execution_options.isolation_level 这些值是字符串，通常是以下名称的子集：

# possible values for Connection.execution_options(isolation_level="<value>")

"AUTOCOMMIT"
"READ COMMITTED"
"READ UNCOMMITTED"
"REPEATABLE READ"
"SERIALIZABLE"

并非每个DBAPI都支持每个值；如果某个后端使用了不受支持的值，则会引发错误。

例如，要在特定连接上强制执行可重复读取，请启动事务：

with engine.connect().execution_options(isolation_level="REPEATABLE READ") as connection:
    with connection.begin():
        connection.execute(<statement>)

这个 Connection.execution_options.isolation_level 选项也可以设置为整个发动机范围，这通常是可取的。这是通过在 create_engine.execution_options 参数到 create_engine() ：：

from sqlalchemy import create_engine

eng = create_engine(
    "postgresql://scott:tiger@localhost/test",
    execution_options={
        "isolation_level": "REPEATABLE READ"
    }
)

使用上述设置，DBAPI连接将被设置为使用 "REPEATABLE READ" 为开始的每个新事务设置隔离级别。

经常选择在不同隔离级别中运行操作的应用程序可能希望创建一个潜在客户的多个“子引擎” Engine ，每个都将配置为不同的隔离级别。一个这样的用例是一个应用程序，它的操作分为“事务性”和“只读”操作，一个单独的 Engine 它利用了 "AUTOCOMMIT" 可与主机分离：

from sqlalchemy import create_engine

eng = create_engine("postgresql://scott:tiger@localhost/test")

autocommit_engine = eng.execution_options(isolation_level="AUTOCOMMIT")

上面， Engine.execution_options() 方法创建原始的浅拷贝 Engine . 两个 eng 和 autocommit_engine 共享相同的方言和连接池。但是，当从获取连接时，“自动提交”模式将在连接上设置 autocommit_engine .

当连接返回到连接池时，将无条件地恢复隔离级别设置（无论是哪一个）。

with engine.connect() as connection:
    connection.execution_options(isolation_level="AUTOCOMMIT")
    connection.execute(<statement>)
    connection.execute(<statement>)

with engine.connect() as connection:
    connection = connection.execution_options(isolation_level="AUTOCOMMIT")

    # this begin() does nothing, isolation stays at AUTOCOMMIT
    with connection.begin() as trans:
        connection.execute(<statement>)
        connection.execute(<statement>)

INFO sqlalchemy.engine.Engine BEGIN (implicit)
...
INFO sqlalchemy.engine.Engine COMMIT using DBAPI connection.commit(), DBAPI should ignore due to autocommit mode

engine = create_engine(..., future=True)

with engine.connect() as connection:
    connection = connection.execution_options(isolation_level="AUTOCOMMIT")

    # "transaction" is autobegin (but has no effect due to autocommit)
    connection.execute(<statement>)

    # this will raise; "transaction" is already begun
    with connection.begin() as trans:
        connection.execute(<statement>)

# if we wanted to flip autocommit on and off on a single connection/
# which... we usually don't.

engine = create_engine(..., future=True)

with engine.connect() as connection:

    connection.execution_options(isolation_level="AUTOCOMMIT")

    # run statement(s) in autocommit mode
    connection.execute(<statement>)

    # "commit" the autobegun "transaction" (2.0/future mode only)
    connection.commit()

    # switch to default isolation level
    connection.execution_options(isolation_level=connection.default_isolation_level)

    # use a begin block
    with connection.begin() as trans:
        connection.execute(<statement>)

engine = create_engine(..., future=True)

# use an autocommit block
with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as connection:

    # run statement in autocommit mode
    connection.execute(<statement>)

# use a regular block
with engine.begin() as connection:
    connection.execute(<statement>)

使用服务器端游标（即流结果）¶

有限数量的方言明确支持“服务器端游标”与“缓冲游标”的概念。虽然在SQLAlchemy的引擎和方言实现中，服务器端游标意味着各种不同的功能，但它只指特定结果集在从游标中提取之前是否在内存中完全缓冲，方法如下 cursor.fetchall() . SQLAlchemy不直接支持游标行为，如滚动；要将这些特性用于特定的DBAPI，请直接使用游标，如中所述 使用驱动程序SQL和原始DBAPI连接 .

有些DBAPI，如cx峎u Oracle DBAPI，只在内部使用服务器端游标。所有结果集在结果集的整个范围内基本上都是无缓冲区的，只利用固定大小的较小缓冲区，例如一次100行。

对于那些对缓冲或无缓冲结果有条件支持的方言，使用“无缓冲”或服务器端游标模式通常有一些警告。例如，当使用psycopg2方言时，如果服务器端游标与任何类型的DML或DDL语句一起使用，则会引发错误。当使用带有服务器端游标的MySQL驱动程序时，DBAPI连接处于更脆弱的状态，无法从错误条件中正常恢复，也不允许在游标完全关闭之前继续回滚。

因此，SQLAlchemy的方言将始终默认为不易出错的游标版本，这意味着对于PostgreSQL和MySQL方言来说，它默认为一个缓冲的“客户端”游标，在从游标调用任何fetch方法之前，会将完整的结果集拉入内存。这种操作模式适用于 绝大多数 通常情况下，无缓冲游标通常不起作用，除非应用程序以块的形式获取大量行，在这种情况下，可以在提取更多行之前完成对这些行的处理。

要在特定执行中使用服务器端游标，则 Connection.execution_options.stream_results 选项，可以在 Connection 对象、语句对象或下面提到的ORM级别上下文中。

当在语句中使用此选项时，通常应该使用如下方法 Result.partitions() 要一次处理结果集的小部分，同时还要为每次拉取获取足够的行，以便操作高效：

with engine.connect() as conn:
    result = conn.execution_options(stream_results=True).execute(text("select * from table"))

    for partition in result.partitions(100):
        _process_rows(partition)

如果 Result 直接迭代，使用默认的缓冲方案在内部获取行，该方案首先缓冲一小部分行，然后在每次提取时缓冲越来越大的缓冲区，达到预先配置的1000行限制。使用 max_row_buffer 执行选项：

with engine.connect() as conn:
    conn = conn.execution_options(stream_results=True, max_row_buffer=100)
    result = conn.execute(text("select * from table"))

    for row in result:
        _process_row(row)

方法将缓冲区的大小设置为固定大小。 Result.yield_per() 方法。使用多行调用此方法将导致所有结果提取方法都从给定大小的缓冲区中工作，仅在缓冲区为空时才提取新行：：

with engine.connect() as conn:
    result = conn.execution_options(stream_results=True).execute(text("select * from table"))

    for row in result.yield_per(100):
        _process_row(row)

这个 stream_results ORM也提供此选项。使用ORM时，可以使用 Result.yield_per() 或 Result.partitions() 方法应用于设置每次产生以下结果时要缓冲的ORM行数：

with orm.Session(engine) as session:
    result = session.execute(
        select(User).order_by(User_id).execution_options(stream_results=True),
    )
    for partition in result.partitions(100):
        _process_rows(partition)

在使用 1.x style ORM查询，使用 Query ，Year_PER可通过以下方式获得 Query.yield_per() -这还会设置 stream_results 执行选项：：

for row in session.query(User).yield_per(100):
    # process row

无连接执行，隐式执行¶

回想一下我们提到的第一节，执行时使用或不使用 Connection . "“无连接”执行是指 execute() 对象上的方法，该对象不是 Connection . 这是用 Engine.execute() 方法 Engine ：：

result = engine.execute(text("select username from users"))
for row in result:
    print("username:", row['username'])

除了“无连接”执行之外，还可以使用 Executable.execute() 任意方法 Executable 构造，它是支持执行的SQL表达式对象的标记。SQL表达式对象本身引用了 Engine 或 Connection 被称为 bind 以提供所谓的“隐式”执行服务。

给出如下表格：

from sqlalchemy import MetaData, Table, Column, Integer

metadata_obj = MetaData()
users_table = Table('users', metadata_obj,
    Column('id', Integer, primary_key=True),
    Column('name', String(50))
)

显式执行将SQL文本或构造的SQL表达式传递到 Connection.execute() 方法 Connection ：

engine = create_engine('sqlite:///file.db')
with engine.connect() as connection:
    result = connection.execute(users_table.select())
    for row in result:
        # ....

显式无连接执行将表达式传递给 Engine.execute() 方法 Engine ：

engine = create_engine('sqlite:///file.db')
result = engine.execute(users_table.select())
for row in result:
    # ....
result.close()

隐式执行也是无连接的，并使用 Executable.execute() 方法。此方法是 Executable 类，它引用的SQL语句足以对数据库调用。该方法使用了以下假设： Engine 或 Connection 已经 跳跃 到表达式对象。我们所说的“绑定”是指 MetaData.bind 已用于关联一系列 Table 对象和所有从这些对象派生的具有特定引擎的SQL构造：：

engine = create_engine('sqlite:///file.db')
metadata_obj.bind = engine
result = users_table.select().execute()
for row in result:
    # ....
result.close()

上面，我们将 Engine 用一个 MetaData 使用特殊属性的对象 MetaData.bind . 这个 select() 从中生成的构造 Table 对象有一个方法 Executable.execute() ，它将搜索 Engine 这是“绑定”到 Table .

总的来说，“绑定元数据”的使用有三个一般影响：

• SQL语句对象获得 Executable.execute() 方法，它自动定位一个“bind”，用它来执行自己。

• 奥姆 Session 对象支持使用“绑定元数据”来确定 Engine 应该用于代表特定的映射类调用SQL语句，尽管 Session 也有其自己的明确的建立复杂系统的特点。 Engine /映射的类配置。

• 这个 MetaData.create_all() ， MetaData.drop_all() ， Table.create() ， Table.drop() 和“autoload”特性都使用绑定 Engine 自动传递，无需显式传递。

在两个“无连接”示例中， Connection 是在幕后创造的；是 CursorResult 返回的 execute() 调用引用 Connection 用于发出SQL语句。当 CursorResult 是关闭的，基础 Connection 为我们关闭，导致DBAPI连接返回到池，事务性资源被删除。

模式名称的翻译¶

要支持将公共表集分布到多个模式中的多租户应用程序，请 Connection.execution_options.schema_translate_map 执行选项可用于重新调整一组 Table 对象以不同的架构名称呈现，不做任何更改。

给出表格：

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

这个的“模式” Table 定义如下： Table.schema 属性是 None . 这个 Connection.execution_options.schema_translate_map 可以全部指定 Table 架构为的对象 None 而是将架构呈现为 user_schema_one ：：

connection = engine.connect().execution_options(
    schema_translate_map={None: "user_schema_one"})

result = connection.execute(user_table.select())

上述代码将在以下表单的数据库上调用SQL:：

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

这个 Connection.execution_options.schema_translate_map 参数影响从SQL表达式语言生成的所有DDL和SQL构造，如从 Table 或 Sequence 物体。它确实 not 影响通过 text() 构造或通过传递给 Connection.execute() .

该功能生效 only 在这些情况下，模式的名称直接从 Table 或 Sequence ；它不会影响直接传递字符串架构名称的方法。根据此模式，它在“can create”/“can drop”检查中生效，这些检查由以下方法执行 MetaData.create_all() 或 MetaData.drop_all() 调用，并在使用表反射时生效 Table 对象。然而它确实如此 not 影响当前的操作 Inspector 对象，因为模式名被显式传递给这些方法。

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

session = Session(schema_engine)

...

SQL编译缓存¶

SQLAlchemy为SQL编译器及其ORM变体提供了一个全面的缓存系统。此缓存系统在 Engine 并规定给定Core或ORM SQL语句的SQL编译过程，以及为该语句装配结果获取机制的相关计算，对于该语句对象和所有其他具有相同结构的对象只会发生一次，在特定结构保留在引擎的“编译缓存”中的持续时间内。通过“具有相同结构的语句对象”，这通常对应于在函数内构造并在每次函数运行时生成的SQL语句：

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)

上面的语句将生成类似于 SELECT id, col FROM table WHERE col = :col ORDER BY id ，注意到 parameter 因为Python是一个纯字符串，或者它不包含字符串形式的字符串，所以它不包含字符串形式的参数。上述内容的后续调用 run_my_statement() 函数将在 connection.execute() 要求提高性能。

虽然SQLAlchemy从1.x系列的早期就已经有了一个基本的语句缓存，并且还为ORM提供了“bakequery”扩展，但是这两个系统都需要大量的特殊API才能使缓存有效。新的缓存从1.4版开始是完全自动的，不需要改变编程风格就可以生效。

缓存是自动使用的，不需要任何配置更改，也不需要特殊步骤来启用它。以下各节详细介绍缓存的配置和高级使用模式。

engine = create_engine("postgresql://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 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.query(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())

from sqlalchemy.engine.default import DefaultDialect

class MyDialect(DefaultDialect):
    supports_statement_cache = True

class MyDBAPIForMyDialect(MyDialect):
    supports_statement_cache = True

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

>>> 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 引用连接池，这意味着在正常情况下，存在打开的数据库连接，而 Engine 对象仍驻留在内存中。当一个 Engine 是垃圾收集的，其连接池不再由该引用 Engine ，并且假设它的所有连接都未签出，那么池及其连接也将被垃圾收集，这也会关闭实际的数据库连接。但除此之外， Engine 将保留打开的数据库连接，假定它使用 QueuePool .

这个 Engine 通常是在应用程序的整个生命周期中预先建立和维护的永久性固定装置。它是 not 目的是在每个连接的基础上创建和处理；相反，它是一个注册中心，既维护连接池，也维护数据库和DBAPI使用中的配置信息，以及每个数据库资源的某种程度的内部缓存。

但是，在许多情况下，需要 Engine 完全封闭。对于这些情况，依赖于Python垃圾收集通常不是一个好主意，而是 Engine 可以使用 Engine.dispose() 方法。这将处理引擎的底层连接池，并将其替换为空的新连接池。前提是 Engine 在此时被丢弃，不再使用 checked-in 它所指的连接也将完全关闭。

调用的有效用例 Engine.dispose() 包括：

• 当一个程序想要释放连接池保留的任何剩余的签入连接，并且期望在将来的任何操作中不再连接到该数据库时。

• 当程序使用多处理或 fork() 和一个 Engine 对象被复制到子进程， Engine.dispose() 应该调用，以便引擎在该分叉的本地创建全新的数据库连接。数据库连接通常是这样的 not 跨越流程边界。

• 在测试套件或多租户场景中，许多临时的、短期的 Engine 可以创建和释放对象。

连接 签出 是 not 当引擎被释放或垃圾收集时丢弃，因为应用程序在其他地方仍然强烈引用这些连接。然而，之后 Engine.dispose() 被调用，这些连接不再与 Engine ；当它们被关闭时，它们将被返回到它们现在的孤立连接池，一旦引用它的所有连接也不再在任何地方被引用，最终将被垃圾收集。由于这一过程不容易控制，强烈建议 Engine.dispose() 只有在签入所有已签出的连接或从其池中取消关联后才调用。

对于受 Engine 对象使用连接池是为了完全禁用池。这通常只会对新连接的使用产生适度的性能影响，这意味着当一个连接被签入时，它将完全关闭，而不会保留在内存中。见 切换池实现 有关如何禁用池的指南。

使用驱动程序SQL和原始DBAPI连接¶

使用简介 Connection.execute() 利用 text() 构造以说明如何调用文本SQL语句。在使用SQLAlchemy时，文本SQL实际上更像是一种例外而不是规范，因为核心表达式语言和ORM都抽象了SQL的文本表示。但是 text() 构造本身还提供了一些文本SQL的抽象，因为它规范了绑定参数的传递方式，并且支持参数和结果集行的数据类型行为。

with engine.connect() as conn:
    conn.exec_driver_sql("SET param='bar'")

connection = engine.connect()
dbapi_conn = connection.connection

dbapi_conn = engine.raw_connection()

dbapi_conn.close()

connection = engine.raw_connection()
try:
    cursor_obj = connection.cursor()
    cursor_obj.callproc("my_procedure", ['x', 'y', 'z'])
    results = list(cursor_obj.fetchall())
    cursor_obj.close()
    connection.commit()
finally:
    connection.close()

connection.execute("CALL my_procedure();")

connection = engine.raw_connection()
try:
    cursor_obj = connection.cursor()
    cursor_obj.execute("select * from table1; select * from table2")
    results_one = cursor_obj.fetchall()
    cursor_obj.nextset()
    results_two = cursor_obj.fetchall()
    cursor_obj.close()
finally:
    connection.close()

注册新方言¶

这个 create_engine() 函数调用使用SETUPTOOLS入口点定位给定方言。可以在setup.py脚本中为第三方方言建立这些入口点。例如，要创建新方言“foodialect://”，步骤如下：

1. 创建名为 foodialect .

2. 包应该有一个包含方言类的模块，该模块通常是 sqlalchemy.engine.default.DefaultDialect . 在这个例子中，假设它被称为 FooDialect 其模块通过 foodialect.dialect .

3. 可以在setup.py中建立入口点，如下所示：

   entry_points="""
   [sqlalchemy.dialects]
   foodialect = foodialect.dialect:FooDialect
   """

如果方言在现有的SQLAlchemy支持的数据库之上为特定DBAPI提供支持，则可以提供名称，包括数据库限定。例如，如果 FooDialect 实际上是一个MySQL方言，入口点可以这样建立：

entry_points="""
[sqlalchemy.dialects]
mysql.foodialect = foodialect.dialect:FooDialect
"""

然后将访问上述入口点作为 create_engine("mysql+foodialect://") .

from sqlalchemy.dialects import registry
registry.register("mysql.foodialect", "myapp.dialect", "MyMySQLDialect")

连接/引擎API¶

class sqlalchemy.engine.Connection(engine, connection=None, close_with_result=False, _branch_from=None, _execution_options=None, _dispatch=None, _has_events=None, _allow_revalidate=True)¶

为打包的DB-API连接提供高级功能。

这是SQLAlChemy 1.x.x版本 的 Connection 班级。对于 2.0 style 版本，该版本具有一些API差异，请参见 Connection 。

这个 Connection 对象是通过调用 Engine.connect() 的方法。 Engine 对象，并为执行SQL语句和事务控制提供服务。

连接对象是 not 线程安全。虽然可以使用适当的同步访问在线程之间共享连接，但底层DBAPI连接仍然可能不支持线程之间的共享访问。有关详细信息，请查看DBAPI文档。

Connection对象表示从连接池签出的单个DBAPI连接。在这种状态下，连接池对连接没有影响，包括其过期或超时状态。为了使连接池能够正确地管理连接，连接应该返回到连接池（即。 connection.close() ）当连接不使用时。

类签名

class sqlalchemy.engine.Connection (sqlalchemy.engine.Connectable)

method sqlalchemy.engine.Connection.__init__(engine, connection=None, close_with_result=False, _branch_from=None, _execution_options=None, _dispatch=None, _has_events=None, _allow_revalidate=True)¶

构建新连接。

method sqlalchemy.engine.Connection.begin()¶

开始事务并返回事务句柄。

返回的对象是 Transaction . 此对象表示事务的“范围”，当 Transaction.rollback() 或 Transaction.commit() 方法被调用。

小技巧

这个 Connection.begin() 方法时调用 Engine.begin() 上下文管理器方法。引用特定于 Connection.begin() 方法也适用于使用 Engine.begin() 方法。

传统使用：嵌套调用 begin() 在同一时间 Connection 将返回新的 Transaction 对象，这些对象表示封闭事务的作用域内的模拟事务，即：

trans = conn.begin()   # outermost transaction
trans2 = conn.begin()  # "nested"
trans2.commit()        # does nothing
trans.commit()         # actually commits

呼叫 Transaction.commit() 只有通过最外层调用时才有效果 Transaction 对象，尽管 Transaction.rollback() 任何一种方法 Transaction 对象将回滚事务。

小技巧

上面的“嵌套”行为是特定于 1.x style 使用和将在SQLAlChemy 2.0中删除。有关以下内容的备注 2.0 style 使用，请参见 Connection.begin() 。

参见

Connection.begin_nested() -使用保存点

Connection.begin_twophase() -使用两阶段/xid事务

Engine.begin() - context manager available from Engine

method sqlalchemy.engine.Connection.begin_nested()¶

假设外部事务已经建立，则开始嵌套事务(即保存点)并返回事务句柄。

嵌套事务需要基础数据库中的保存点支持。层次结构中的任何事务都可以 commit 和 rollback 但是，最外层的事务仍然控制整个 commit 或 rollback 一个整体的交易。

The legacy form of Connection.begin_nested() method has alternate behaviors based on whether or not the Connection.begin() method was called previously. If Connection.begin() was not called, then this method will behave the same as the Connection.begin() method and return a RootTransaction object that begins and commits a real transaction - no savepoint is invoked. If Connection.begin() has been called, and a RootTransaction is already established, then this method returns an instance of NestedTransaction which will invoke and manage the scope of a SAVEPOINT.

小技巧

上述行为 Connection.begin_nested() 是特定于 1.x style 使用。在……里面 2.0 style 使用、 Connection.begin_nested() 方法而是自动开始可以使用“随用随用”方式提交的外部事务；请参见 Connection.begin_nested() 有关迁移详细信息，请参阅。

在 1.4.13 版更改: 的行为 Connection.begin_nested() 作为返回一个 RootTransaction 如果 Connection.begin() 在1.3.x版本中是这样的；在以前的1.4.x版本中，外部事务将是“自动开始的”，但不会被提交。

参见

Connection.begin()

Connection.begin_twophase()

method sqlalchemy.engine.Connection.begin_twophase(xid=None)¶

开始两阶段或XA事务并返回事务句柄。

返回的对象是 TwoPhaseTransaction 除了提供的方法外， Transaction ，还提供 TwoPhaseTransaction.prepare() 方法。

参数

xid¶ -- 两阶段事务ID。如果未提供，将生成随机ID。

参见

Connection.begin()

Connection.begin_twophase()

method sqlalchemy.engine.Connection.close()¶

关闭此 Connection .

这将导致底层数据库资源的发布，即内部引用的DBAPI连接。DBAPI连接通常恢复到保持连接的状态。 Pool 被引用 Engine 产生了这个 Connection . DBAPI连接上存在的任何事务状态也会通过DBAPI连接无条件释放 rollback() 方法，无论 Transaction 与此相关的未解决对象 Connection .

后 Connection.close() 被称为 Connection 永久处于关闭状态，不允许进一步操作。

attribute sqlalchemy.engine.Connection.closed¶

如果此连接已关闭，则返回true。

method sqlalchemy.engine.Connection.connect(close_with_result=False)¶

返回此的分支版本 Connection .

1.4 版后已移除: 这个 Connection.connect() 方法在SQLAlchemy的1.x系列中被认为是遗留的，并将在2.0中删除。（SQLAlchemy 2.0的背景： 迁移到Alchemy ）

这个 Connection.close() 返回的方法 Connection 可以叫这个 Connection 将保持打开状态。

这种方法提供了使用对称性 Engine.connect() 包括用于上下文管理器。

attribute sqlalchemy.engine.Connection.connection¶

此连接管理的基础DB-API连接。

这是SQLAlChemy连接池代理连接，然后该连接具有属性 _ConnectionFairy.dbapi_connection 这指的是实际的驱动程序连接。

参见

使用驱动程序SQL和原始DBAPI连接

attribute sqlalchemy.engine.Connection.default_isolation_level¶

分配给此的默认隔离级别 Connection .

这是隔离级别设置， Connection 第一次通过 Engine.connect() 方法。此级别在 Connection.execution_options.isolation_level 用于更改PER上的设置 -Connection 基础。

不像 Connection.get_isolation_level() ，此属性是从方言获取的第一个连接提前设置的，因此在调用此访问器时不会调用SQL查询。

0.9.9 新版功能.

参见

Connection.get_isolation_level() -查看当前级别

create_engine.isolation_level -设置 Engine 隔离级别

Connection.execution_options.isolation_level -设置 Connection 隔离级别

method sqlalchemy.engine.Connection.detach()¶

将基础DB-API连接与其连接池分离。

例如。：：

with engine.connect() as conn:
    conn.detach()
    conn.execute(text("SET search_path TO schema1, schema2"))

    # work with connection

# connection is fully closed (since we used "with:", can
# also call .close())

这个 Connection 实例将保持可用。当关闭（或从如上所述的上下文管理器上下文退出）时，DB-API连接将完全关闭，而不会返回到其原始池。

此方法可用于将应用程序的其余部分与连接上的修改状态隔离（例如事务隔离级别或类似级别）。

method sqlalchemy.engine.Connection.exec_driver_sql(statement, parameters=None, execution_options=None)¶

执行SQL语句构造并返回 CursorResult .

参数

• statement¶ -- 要执行的语句str。绑定参数必须使用底层DBAPI的paramstyle，例如“qmark”、“pyformat”、“format”等。

• parameters¶ -- 表示要在执行中使用的绑定参数值。该格式是以下格式之一：命名参数的字典、位置参数的元组，或包含字典或元组的列表，以支持多个执行。E、 g.多个词典：conn.exec_驱动程序u sql（“插入表（id，value）值（%（id）s，%（value）s）”， [{“id”：1，“value”：“v1”}}，{“id”：2，“value”：“v2”}] )单个词典：conn.exec_驱动程序u sql（“插入到表（id，value）值（%（id）s，%（value）s）”，dict（id=1，value=“v1”））单元组：：conn.exec_驱动程序u sql（“插入表（id，value）值（？）？, ?)"，（1，'v1'））。。注意：： Connection.exec_driver_sql() 方法不参与 ConnectionEvents.before_execute() 和 ConnectionEvents.after_execute() 事件。拦截呼叫 Connection.exec_driver_sql() 使用 ConnectionEvents.before_cursor_execute() 和 ConnectionEvents.after_cursor_execute() . …参阅： PEP 249

method sqlalchemy.engine.Connection.execute(statement, *multiparams, **params)¶

执行SQL语句构造并返回 CursorResult .

参数

• statement¶ -- 要执行的语句。可能是： 纯字符串（已弃用） 任何 ClauseElement 构造，它也是 Executable ，比如 select() 建造 * a FunctionElement, such as that generated by func, will be automatically wrapped in a SELECT statement, which is then executed. * 一 DDLElement 对象 * a DefaultGenerator object * 一 Compiled 对象。。已弃用：：2.0正在将字符串传递给 Connection.execute() 已弃用，将在版本2.0中删除。使用 text() 构建与 Connection.execute() ，或者 Connection.exec_driver_sql() 方法调用驱动程序级别的SQL字符串。

• *multiparams/**params¶ -- 表示要在执行中使用的绑定参数值。通常，格式是传递给的一个或多个字典的集合 *multiparams:: conn.execute( table.insert(), {{"id":1, "value":"v1"}}, {{"id":2, "value":"v2"}} ) ...or individual key/values interpreted by * *params:: conn.execute( table.insert(), id=1, value="v1" ) In the case that a plain SQL string is passed, and the underlying DBAPI accepts positional bind parameters, a collection of tuples or individual values in * 可以传递multiparams：：Conn.ecute(“INSERT INTO TABLE(id，VALUE)VALUES(？，？)”，(1，“v1”)，(2，“v2”))CONNEXTE(“INSERT INTO TABLE(id，VALUE)VALUES(？，？)”，1，“v1”)以上注意事项，问号的用法“？”或其他符号取决于所使用的DBAPI接受的“paramstyle”，其可以是“qmark”、“named”、“pyformat”、“format”、“numeric”中的任何一个。看见 pep-249 有关参数样式的详细信息，请参见。若要以与DBAPI无关的方式执行使用绑定参数的文本SQL语句，请使用 text() 构造。。。已弃用：：2.0不推荐使用元组或标量位置参数。所有参数都应该是DICT或DICT序列。使用 exec_driver_sql() 执行带有元组或标量位置参数的纯字符串。

method sqlalchemy.engine.Connection.execution_options(**opt)¶

为在执行期间生效的连接设置非SQL选项。

对于“future”样式的连接，此方法返回相同的 Connection 对象添加了新选项。

对于旧连接，此方法返回 Connection 它引用相同的基础DBAPI连接，但也定义了给定的执行选项，这些选项将在调用 execute() . 作为新的 Connection 引用相同的基础资源，通常最好确保立即丢弃副本，这在以下情况下是隐式的：

result = connection.execution_options(stream_results=True).\
                    execute(stmt)

注意，任何键/值都可以传递给 Connection.execution_options() ，它将存储在 _execution_options 字典 Connection . 例如，它适合于最终用户方案与事件侦听器通信。

当前由sqlacalchemy本身识别的关键字包括下面列出的所有关键字 Executable.execution_options() 以及其他特定于 Connection .

参数

• autocommit¶ -- 可用于：连接，语句。如果为True，则在“autocommit”模式下执行时，即在连接上没有开始显式事务时，将在执行后调用提交。注意这是 库级别，而不是DBAPI级别的自动提交 . 除非使用“AUTOCOMMIT”隔离级别，否则DBAPI连接将保留在实际事务中。。已弃用：：1.4“autocommit”执行选项已弃用，将在SQLAlchemy 2.0中删除。看到了吗 库级（但不是驱动程序级）“自动提交”从核心和ORM中删除 供讨论。

• compiled_cache¶ -- 在：连接上可用。字典在哪里 Compiled 当 Connection 将子句表达式编译为 Compiled 对象。此字典将取代在上配置的语句缓存 Engine 本身。如果设置为“无”，则禁用缓存，即使引擎已配置缓存大小。注意，它自己的一些操作，包括刷新操作。ORM内部使用的缓存将取代此处指定的缓存字典。

• logging_token¶ -- 可在以下位置获得： Connection ， Engine 。在连接记录的日志消息中添加由括号括起来的指定字符串标记，即通过 create_engine.echo 标志或通过 logging.getLogger("sqlalchemy.engine") 伐木者。这允许每个连接或每个子引擎的令牌可用，这对于调试并发连接方案很有用。。。版本添加：：1.4.0b2..另请参阅：： 设置每个连接/子引擎令牌 -使用示例 create_engine.logging_name -将名称添加到Python记录器对象本身使用的名称。

• isolation_level¶ --

  可供选择： Connection .

  为此的生命周期设置事务隔离级别 Connection 对象。有效值包括 create_engine.isolation_level 传递给的参数 create_engine() . 这些级别是半数据库特定的；有关有效级别，请参阅各个方言文档。

  隔离级别选项通过在DBAPI连接上发出语句来应用隔离级别，并且 必然会整体影响原始连接对象 ，而不仅仅是调用返回的副本 Connection.execution_options() 方法。隔离级别将保持在给定的设置，直到DBAPI连接本身返回到连接池，即 Connection.close() 原方法 Connection 调用时，事件处理程序将在DBAPI连接上发出附加语句，以便恢复隔离级别更改。

  警告

  这个 isolation_level 执行选项应该 not 当事务已经建立时使用，即 Connection.begin() 已调用方法或类似方法。数据库无法更改正在进行的事务的隔离级别，不同的dbapis和/或sqlachemy方言可能隐式回滚或提交事务，或者根本不影响连接。

  注解

  这个 isolation_level 如果 Connection 无效，例如通过 Connection.invalidate() 方法，或者如果发生断开错误。失效后产生的新连接不会自动重新应用隔离级别。

  参见

  create_engine.isolation_level -设置 Engine 隔离级别

  Connection.get_isolation_level() -查看当前级别

  SQLite Transaction Isolation

  PostgreSQL Transaction Isolation

  MySQL Transaction Isolation

  SQL Server Transaction Isolation

  设置事务隔离级别/DBAPI AUTOCOMMIT -为ORM

• no_parameters¶ -- 什么时候？ True 如果最终参数列表或字典完全为空，则将调用光标上的语句 cursor.execute(statement) ，根本不传递参数集合。一些DBAPI（如psycopg2和mysql python）仅在存在参数时才将百分号视为重要的符号；此选项允许代码生成包含百分号（可能还有其他字符）的SQL，这些百分号与DBAPI是否执行它或是否通过管道传输到稍后由命令行工具调用的脚本无关。

• stream_results¶ -- 可用于：连接，语句。如果可能的话，告诉方言结果应该是“流式”的，而不是预缓冲的。这是许多dbapi的一个限制。该标志目前在PostgreSQL和MySQL类别的方言子集中被理解，并且可能被其他第三方方言所支持。。另请参见： 使用服务器端游标（即流结果）

• schema_translate_map¶ -- 提供：连接，发动机。将架构名称映射到架构名称的字典，该字典将应用于 Table.schema 各元素 Table 当SQL或DDL表达式元素编译为字符串时遇到；将根据原始名称的映射中的存在来转换生成的架构名称。…版本已添加：：1.1..参阅： 模式名称的翻译

参见

Engine.execution_options()

Executable.execution_options()

Connection.get_execution_options()

method sqlalchemy.engine.Connection.get_execution_options()¶

获取将在执行期间生效的非SQL选项。

1.3 新版功能.

参见

Connection.execution_options()

method sqlalchemy.engine.Connection.get_isolation_level()¶

返回分配给此的当前隔离级别 Connection .

这通常是由方言确定的默认隔离级别，除非 Connection.execution_options.isolation_level 已使用功能更改每个上的隔离级别 -Connection 基础。

此属性通常会执行实时SQL操作以获取当前隔离级别，因此返回的值是基础DBAPI连接的实际级别，而不管如何设置此状态。相比于 Connection.default_isolation_level 访问器，返回方言级别设置而不执行SQL查询。

0.9.9 新版功能.

参见

Connection.default_isolation_level -查看默认级别

create_engine.isolation_level -设置 Engine 隔离级别

Connection.execution_options.isolation_level -设置 Connection 隔离级别

method sqlalchemy.engine.Connection.get_nested_transaction()¶

返回当前正在进行的嵌套事务（如果有）。

1.4 新版功能.

method sqlalchemy.engine.Connection.get_transaction()¶

返回当前正在进行的根事务（如果有）。

1.4 新版功能.

method sqlalchemy.engine.Connection.in_nested_transaction()¶

如果事务正在进行，则返回true。

method sqlalchemy.engine.Connection.in_transaction()¶

如果事务正在进行，则返回true。

attribute sqlalchemy.engine.Connection.info¶

与此引用的基础DBAPI连接关联的信息字典 Connection ，允许用户定义的数据与连接关联。

这里的数据将与DBAPI连接一起使用，包括在它返回到连接池并在随后的实例中再次使用之后。 Connection .

method sqlalchemy.engine.Connection.invalidate(exception=None)¶

使与此关联的基础DBAPI连接失效 Connection .

将尝试立即关闭底层DBAPI连接；但是，如果此操作失败，将记录错误，但不会引发错误。无论close（）是否成功，连接都将被丢弃。

下次使用时（其中“使用”通常指使用 Connection.execute() 方法或类似方法），此 Connection 将尝试使用 Pool 作为连接源（例如“重新连接”）。

如果交易正在进行（例如 Connection.begin() 方法）当 Connection.invalidate() 方法在dbapi级别被调用，当dbapi连接关闭时，与此事务关联的所有状态都将丢失。这个 Connection 将不允许重新连接继续，直到 Transaction 对象结束，方法是调用 Transaction.rollback() 方法；在此之前，继续使用 Connection 将提高 InvalidRequestError . 这是为了防止应用程序意外地继续正在进行的事务操作，尽管事务由于无效而丢失。

这个 Connection.invalidate() 方法和自动失效一样，将在连接池级别调用 PoolEvents.invalidate() 事件。

参数

exception¶ -- 可选的 Exception 实例这就是无效的原因。传递给事件处理程序和日志记录函数。

参见

关于无效的更多信息

attribute sqlalchemy.engine.Connection.invalidated¶

如果此连接无效，则返回true。

method sqlalchemy.engine.Connection.run_callable(callable_, *args, **kwargs)¶

给定一个可调用的对象或函数，执行它，传递 Connection 作为第一个论点。

1.4 版后已移除: 这个 Connection.run_callable() 方法已弃用，并将在将来的版本中删除。传递连接，直接调用可调用函数。

给定的 * ARGS和 * *Kwargs在 Connection 争论。

这个函数，以及 Engine.run_callable() ，允许使用 Connection 或 Engine 对象，不需要知道正在处理哪个对象。

method sqlalchemy.engine.Connection.scalar(object_, *multiparams, **params)¶

执行并返回第一行的第一列。

基础结果/光标在执行后关闭。

method sqlalchemy.engine.Connection.scalars(object_, *multiparams, **params)¶

执行并返回标量结果集，该结果集从每行的第一列生成标量值。

此方法等效于调用 Connection.execute() 要获得一个 Result 对象，然后调用 Result.scalars() 方法来生成 ScalarResult 实例。

返回

一个 ScalarResult

1.4.24 新版功能.

method sqlalchemy.engine.Connection.schema_for_object(obj)¶

返回给定架构项的架构名称，同时考虑当前架构转换映射。

method sqlalchemy.engine.Connection.transaction(callable_, *args, **kwargs)¶

在事务边界内执行给定的函数。

1.4 版后已移除: 这个 Connection.transaction() 方法已弃用，将在将来的版本中删除。使用 Engine.begin() 而不是上下文管理器。

函数通过此 Connection 作为第一个参数，后跟给定的 * ARGS和 * ＊KWARG，例如：

def do_something(conn, x, y):
    conn.execute(text("some statement"), {'x':x, 'y':y})

conn.transaction(do_something, 5, 10)

函数内部的操作都是在单个 Transaction . 一旦成功，事务就被提交。如果引发异常，则在传播异常之前回滚事务。

注解

这个 transaction() 方法被python的用法取代 with: 语句，可用于 Connection.begin() ：：

with conn.begin():
    conn.execute(text("some statement"), {'x':5, 'y':10})

以及与 Engine.begin() ：：

with engine.begin() as conn:
    conn.execute(text("some statement"), {'x':5, 'y':10})

参见

Engine.begin() -引擎级事务上下文

Engine.transaction() - engine-level version of Connection.transaction()

class sqlalchemy.engine.CreateEnginePlugin(url, kwargs)¶

一组钩子，用于增强 Engine 基于URL中入口点名称的对象。

目的 CreateEnginePlugin 允许第三方系统应用引擎、池和方言级别的事件侦听器，而不需要修改目标应用程序；相反，插件名称可以添加到数据库URL。目标应用程序 CreateEnginePlugin 包括：

• 连接和SQL性能工具，例如使用事件跟踪签出次数和/或语句花费的时间

• 连接插件，如代理

雏形 CreateEnginePlugin 将一个记录器连接到 Engine 对象可能看起来像：

import logging

from sqlalchemy.engine import CreateEnginePlugin
from sqlalchemy import event

class LogCursorEventsPlugin(CreateEnginePlugin):
    def __init__(self, url, kwargs):
        # consume the parameter "log_cursor_logging_name" from the
        # URL query
        logging_name = url.query.get("log_cursor_logging_name", "log_cursor")

        self.log = logging.getLogger(logging_name)

    def update_url(self, url):
        "update the URL to one that no longer includes our parameters"
        return url.difference_update_query(["log_cursor_logging_name"])

    def engine_created(self, engine):
        "attach an event listener after the new Engine is constructed"
        event.listen(engine, "before_cursor_execute", self._log_event)


    def _log_event(
        self,
        conn,
        cursor,
        statement,
        parameters,
        context,
        executemany):

        self.log.info("Plugin logged cursor event: %s", statement)

插件的注册使用入口点的方式与方言类似：

entry_points={
    'sqlalchemy.plugins': [
        'log_cursor_plugin = myapp.plugins:LogCursorEventsPlugin'
    ]

使用上述名称的插件将从数据库URL调用，如下所示：

from sqlalchemy import create_engine

engine = create_engine(
    "mysql+pymysql://scott:tiger@localhost/test?"
    "plugin=log_cursor_plugin&log_cursor_logging_name=mylogger"
)

这个 plugin URL参数支持多个实例，因此一个URL可以指定多个插件；它们是按照URL中声明的顺序加载的：

engine = create_engine(
  "mysql+pymysql://scott:tiger@localhost/test?"
  "plugin=plugin_one&plugin=plugin_twp&plugin=plugin_three")

插件名称也可以直接传递给 create_engine() 使用 create_engine.plugins 论点：

engine = create_engine(
  "mysql+pymysql://scott:tiger@localhost/test",
  plugins=["myplugin"])

1.2.3 新版功能: 插件名称也可以指定为 create_engine() 作为清单

插件可以使用来自 URL 对象以及 kwargs 字典，它是传递给 create_engine() 打电话。”消费”这些参数包括当插件初始化时必须删除它们，这样参数就不会传递给 Dialect 构造函数，在那里他们将引发一个 ArgumentError 因为他们不知道方言。

从SQLAlchemy的1.4版本开始，参数应该继续从 kwargs 通过使用诸如 dict.pop . 来自 URL 对象的 CreateEnginePlugin.update_url() 方法，返回 URL 移除了特定参数的插件：

class MyPlugin(CreateEnginePlugin):
    def __init__(self, url, kwargs):
        self.my_argument_one = url.query['my_argument_one']
        self.my_argument_two = url.query['my_argument_two']
        self.my_argument_three = kwargs.pop('my_argument_three', None)

    def update_url(self, url):
        return url.difference_update_query(
            ["my_argument_one", "my_argument_two"]
        )

上面所示的参数将从 create_engine() 例如：

from sqlalchemy import create_engine

engine = create_engine(
  "mysql+pymysql://scott:tiger@localhost/test?"
  "plugin=myplugin&my_argument_one=foo&my_argument_two=bar",
  my_argument_three='bat'
)

在 1.4 版更改: 这个 URL 对象现在是不可变的；a CreateEnginePlugin 需要改变 URL 应实施新增 CreateEnginePlugin.update_url() 方法，在构建插件后调用。

对于迁移，按照以下方式构造插件，检查是否存在 CreateEnginePlugin.update_url() 方法来检测哪个版本正在运行：

class MyPlugin(CreateEnginePlugin):
    def __init__(self, url, kwargs):
        if hasattr(CreateEnginePlugin, "update_url"):
            # detect the 1.4 API
            self.my_argument_one = url.query['my_argument_one']
            self.my_argument_two = url.query['my_argument_two']
        else:
            # detect the 1.3 and earlier API - mutate the
            # URL directly
            self.my_argument_one = url.query.pop('my_argument_one')
            self.my_argument_two = url.query.pop('my_argument_two')

        self.my_argument_three = kwargs.pop('my_argument_three', None)

    def update_url(self, url):
        # this method is only called in the 1.4 version
        return url.difference_update_query(
            ["my_argument_one", "my_argument_two"]
        )

参见

URL对象现在是不可变的 -概述 URL 变更还包括关于 CreateEnginePlugin .

当引擎创建过程完成并生成 Engine 对象，它再次通过 CreateEnginePlugin.engine_created() 钩子。在这个钩子中，可以对引擎进行额外的更改，最典型的是涉及事件设置（例如，在 核心事件 ）

1.1 新版功能.

method sqlalchemy.engine.CreateEnginePlugin.__init__(url, kwargs)¶

构建新的 CreateEnginePlugin .

插件对象在每次调用 create_engine() . 单一的 Engine 将传递给 CreateEnginePlugin.engine_created() 与此URL对应的方法。

参数

• url¶ -- 这个 URL 对象。该插件可以检查 URL 作为论据。应该删除插件使用的参数，方法是返回更新的 URL 从 CreateEnginePlugin.update_url() 方法。。。版本已更改：：1.4 URL 对象现在是不可变的，因此 CreateEnginePlugin 这需要更改 URL 对象应实现 CreateEnginePlugin.update_url() 方法。

• kwargs¶ -- 传递给的关键字参数 create_engine() .

method sqlalchemy.engine.CreateEnginePlugin.engine_created(engine)¶

接收 Engine 对象完全构造后。

插件可以对引擎进行其他更改，例如注册引擎或连接池事件。

method sqlalchemy.engine.CreateEnginePlugin.handle_dialect_kwargs(dialect_cls, dialect_args)¶

解析和修改方言Kwargs

method sqlalchemy.engine.CreateEnginePlugin.handle_pool_kwargs(pool_cls, pool_args)¶

解析和修改池Kwargs

method sqlalchemy.engine.CreateEnginePlugin.update_url(url)¶

更新 URL .

一个新的 URL 应该归还。此方法通常用于使用来自 URL 必须删除，因为他们不会被方言识别。这个 URL.difference_update_query() 方法可用于删除这些参数。查看docstring CreateEnginePlugin 举个例子。

1.4 新版功能.

class sqlalchemy.engine.Engine(pool, dialect, url, logging_name=None, echo=None, query_cache_size=500, execution_options=None, hide_parameters=False)¶

连接一个 Pool 和 Dialect 一起提供数据库连接和行为的源。

这是 SQLAlChemy 1.x版 的 Engine 。对于 2.0 style 版本，其中包括一些API差异，请参见 Engine 。

安 Engine 对象是使用 create_engine() 功能。

参见

引擎配置

使用引擎和接头

类签名

class sqlalchemy.engine.Engine (sqlalchemy.engine.Connectable, sqlalchemy.log.Identified)

method sqlalchemy.engine.Engine.begin(close_with_result=False)¶

返回一个上下文管理器 Connection 用一个 Transaction 建立。

例如。：：

with engine.begin() as conn:
    conn.execute(
        text("insert into table (x, y, z) values (1, 2, 3)")
    )
    conn.execute(text("my_special_procedure(5)"))

操作成功后， Transaction 承诺。如果出现错误，则 Transaction 回滚。

仅限传统使用： close_with_result 标志通常是 False ，并指示 Connection 将在操作完成后关闭。当设置为时 True ，它表示 Connection 处于“单机使用”模式，其中 CursorResult 的第一次调用返回 Connection.execute() 将关闭 Connection 当那件事发生的时候 CursorResult 已用尽所有结果行。

参见

Engine.connect() -获得A Connection 从一个 Engine .

Connection.begin() -启动一个 Transaction 对于一个特定的 Connection .

method sqlalchemy.engine.Engine.clear_compiled_cache()¶

清除与方言关联的已编译缓存。

这适用 only 到通过建立的内置缓存 create_engine.query_cache_size 参数。它不会影响通过 Connection.execution_options.query_cache 参数。

1.4 新版功能.

method sqlalchemy.engine.Engine.connect(close_with_result=False)¶

返回一个新的 Connection 对象。

这个 Connection 对象是在内部使用DBAPI连接以便与数据库通信的外观。此连接是从连接保持中获得的 Pool 引用人： Engine . 当 Connection.close() 方法 Connection 对象，然后将基础DBAPI连接返回到连接池，在随后的对 Engine.connect() .

method sqlalchemy.engine.Engine.dispose()¶

释放此使用的连接池 Engine .

这有完全关闭所有 当前已签入 数据库连接。仍被签出的连接将 not 关闭，但它们将不再与此关联 Engine 因此，当它们单独关闭时，最终 Pool 它们与之相关联的将被垃圾收集，并且如果在签入时尚未关闭，它们将被完全关闭。

在释放旧连接池后立即创建新连接池。与所有SQLAlchemy连接池一样，只要 Engine 不再使用，不会建立新连接。

参见

引擎处理

attribute sqlalchemy.engine.Engine.driver¶

驱动程序名称 Dialect 在使用中 Engine .

method sqlalchemy.engine.Engine.execute(statement, *multiparams, **params)¶

执行给定的构造并返回 CursorResult .

1.4 版后已移除: 这个 Engine.execute() 方法在SQLAlchemy的1.x系列中被认为是遗留的，并将在2.0中删除。SQLAlchemy 2.0中的所有语句执行都由 Connection.execute() 方法 Connection ，或在ORM中 Session.execute() 方法 Session . （SQLAlchemy 2.0的背景： 迁移到Alchemy ）

这些参数与 Connection.execute() .

这里，A Connection 是使用 Engine.connect() 方法，以及使用该连接执行的语句。归还的人 CursorResult 标记为当 CursorResult 已耗尽并且其基础光标已关闭， Connection 在此创建的数据库也将关闭，这将允许其关联的DBAPI连接资源返回到连接池。

method sqlalchemy.engine.Engine.execution_options(**opt)¶

返回一个新的 Engine 这将提供 Connection 具有给定执行选项的对象。

归还的人 Engine 与原件有关 Engine 因为它共享相同的连接池和其他状态：

• 这个 Pool 新使用 Engine 是同一个实例。这个 Engine.dispose() 方法将替换父引擎以及此引擎的连接池实例。

• 事件侦听器是“级联”的，也就是说，新的 Engine 继承父级的事件，新事件可以与新事件关联 Engine 个别地。

• 日志配置和日志记录名称从父级复制 Engine .

的意图 Engine.execution_options() 方法是在多个 Engine 对象引用相同的连接池，但由自定义事件将使用的选项区分：

primary_engine = create_engine("mysql://")
shard1 = primary_engine.execution_options(shard_id="shard1")
shard2 = primary_engine.execution_options(shard_id="shard2")

上面， shard1 发动机作为工厂 Connection 将包含执行选项的对象 shard_id=shard1 和 shard2 将生产 Connection 包含执行选项的对象 shard_id=shard2 .

事件处理程序可以使用上述执行选项在给定连接的情况下执行模式切换或其他操作。下面我们发出一个mysql use 语句切换数据库，同时跟踪使用 Connection.info 字典，它为我们提供了遵循DBAPI连接的持久存储空间：

from sqlalchemy import event
from sqlalchemy.engine import Engine

shards = {"default": "base", shard_1: "db1", "shard_2": "db2"}

@event.listens_for(Engine, "before_cursor_execute")
def _switch_shard(conn, cursor, stmt,
        params, context, executemany):
    shard_id = conn._execution_options.get('shard_id', "default")
    current_shard = conn.info.get("current_shard", None)

    if current_shard != shard_id:
        cursor.execute("use %s" % shards[shard_id])
        conn.info["current_shard"] = shard_id

参见

Connection.execution_options() -更新上的执行选项 Connection 对象。

Engine.update_execution_options() -更新给定的执行选项 Engine 就位。

Engine.get_execution_options()

method sqlalchemy.engine.Engine.get_execution_options()¶

获取将在执行期间生效的非SQL选项。

参见

Engine.execution_options()

method sqlalchemy.engine.Engine.has_table(table_name, schema=None)¶

如果给定后端具有给定名称的表，则返回true。

1.4 版后已移除: 这个 Engine.has_table() 方法已弃用，将在将来的版本中删除。请参考 Inspector.has_table() .

参见

带检查员的细颗粒反射 -使用 Inspector 接口。

quoted_name -用于传递引用信息和模式标识符。

attribute sqlalchemy.engine.Engine.name¶

字符串名称 Dialect 在使用中 Engine .

method sqlalchemy.engine.Engine.raw_connection(_connection=None)¶

从连接池返回“原始”DBAPI连接。

返回的对象是正在使用的底层驱动程序所使用的DBAPI连接对象的代理版本。对象将具有与实际DBAPI连接相同的行为，除了 close() 方法将导致连接返回池，而不是实际关闭。

当API由提供时，此方法为特殊情况提供直接DBAPI连接访问 Connection 不需要。当A Connection 对象已存在，DBAPI连接可使用 Connection.connection 访问器。

参见

使用驱动程序SQL和原始DBAPI连接

method sqlalchemy.engine.Engine.run_callable(callable_, *args, **kwargs)¶

给定一个可调用的对象或函数，执行它，传递 Connection 作为第一个论点。

1.4 版后已移除: 这个 Engine.run_callable() 方法已弃用，并将在将来的版本中删除。使用 Engine.begin() 取而代之的是上下文管理器。

给定的 * ARGS和 * *Kwargs在 Connection 争论。

这个函数，以及 Connection.run_callable() ，允许使用 Connection 或 Engine 对象，不需要知道正在处理哪个对象。

method sqlalchemy.engine.Engine.scalar(statement, *multiparams, **params)¶

执行并返回第一行的第一列。

1.4 版后已移除: 这个 Engine.scalar() 方法在SQLAlchemy的1.x系列中被认为是遗留的，并将在2.0中删除。SQLAlchemy 2.0中的所有语句执行都由 Connection.execute() 方法 Connection ，或在ORM中 Session.execute() 方法 Session ； Result.scalar() 方法可用于返回标量结果。（SQLAlchemy 2.0的背景： 迁移到Alchemy ）

基础结果/光标在执行后关闭。

method sqlalchemy.engine.Engine.table_names(schema=None, connection=None)¶

返回数据库中所有可用表名的列表。

1.4 版后已移除: 这个 Engine.table_names() 方法已弃用，将在将来的版本中删除。请参考 Inspector.get_table_names() .

参数

• schema¶ -- 可选，从非默认架构中检索名称。

• connection¶ -- 可选，使用指定的连接。

method sqlalchemy.engine.Engine.transaction(callable_, *args, **kwargs)¶

在事务边界内执行给定的函数。

1.4 版后已移除: 这个 Engine.transaction() 方法已弃用，将在将来的版本中删除。使用 Engine.begin() 而不是上下文管理器。

函数通过 Connection 新购自 Engine.connect() 作为第一个参数，后跟给定的 * ARGS和 * *克沃斯。

例如。：：

def do_something(conn, x, y):
    conn.execute(text("some statement"), {'x':x, 'y':y})

engine.transaction(do_something, 5, 10)

函数内部的操作都是在单个 Transaction . 一旦成功，事务就被提交。如果引发异常，则在传播异常之前回滚事务。

注解

这个 transaction() 方法被python的用法取代 with: 语句，可用于 Engine.begin() ：：

with engine.begin() as conn:
    conn.execute(text("some statement"), {'x':5, 'y':10})

参见

Engine.begin() -引擎级事务上下文

Connection.transaction() - connection-level version of Engine.transaction()

method sqlalchemy.engine.Engine.update_execution_options(**opt)¶

更新此的默认执行选项字典 Engine .

中给定的键/值 * *opt被添加到将用于所有连接的默认执行选项中。本词典的初始内容可通过 execution_options 参数到 create_engine() .

参见

Connection.execution_options()

Engine.execution_options()