异步 I/O (asyncio)

Asynchronous I/O (asyncio)

中文

对 Python asyncio 的支持。支持 Core 和 ORM 使用，使用 asyncio 兼容的方言。

在 1.4 版本加入.

警告

请阅读 Asyncio 平台安装说明 以获取有关 所有 平台的重要安装说明。

参见

Asynchronous IO Support for Core and ORM - 初始功能公告

Asyncio 集成 - 示例脚本，展示在 asyncio 扩展中使用 Core 和 ORM 的工作示例。

英文

Support for Python asyncio. Support for Core and ORM usage is included, using asyncio-compatible dialects.

在 1.4 版本加入.

警告

Please read Asyncio 平台安装说明 for important platform installation notes on all platforms.

参见

Asynchronous IO Support for Core and ORM - initial feature announcement

Asyncio 集成 - example scripts illustrating working examples of Core and ORM use within the asyncio extension.

Asyncio 平台安装说明

Asyncio Platform Installation Notes

中文

asyncio 扩展依赖于 greenlet 库。这个依赖 默认未安装。

要安装 SQLAlchemy 并确保存在 greenlet 依赖项，可以安装 [asyncio] setuptools extra，如下所示，这还将包括指示 pip 安装 greenlet：

pip install sqlalchemy[asyncio]

注意，在没有预构建 wheel 文件的平台上安装 greenlet 意味着 greenlet 将从源代码构建，这要求 Python 的开发库也必须存在。

在 2.1 版本发生变更: greenlet 不再默认安装；要使用 asyncio 扩展，必须使用 sqlalchemy[asyncio] 目标。

英文

The asyncio extension depends upon the greenlet library. This dependency is not installed by default.

To install SQLAlchemy while ensuring the greenlet dependency is present, the [asyncio] setuptools extra may be installed as follows, which will include also instruct pip to install greenlet:

pip install sqlalchemy[asyncio]

Note that installation of greenlet on platforms that do not have a pre-built wheel file means that greenlet will be built from source, which requires that Python’s development libraries also be present.

在 2.1 版本发生变更: greenlet is no longer installed by default; to use the asyncio extension, the sqlalchemy[asyncio] target must be used.

概要 - Core

Synopsis - Core

中文

对于 Core 使用，create_async_engine() 函数会创建一个 AsyncEngine 的实例，它提供了传统 Engine API 的异步版本。 AsyncEngine 通过其 AsyncEngine.connect() 和 AsyncEngine.begin() 方法提供 AsyncConnection，这两个方法 都是异步上下文管理器。随后 AsyncConnection 可使用 AsyncConnection.execute() 方法来执行语句并返回一个缓冲的 Result，或使用 AsyncConnection.stream() 方法返回一个 服务器端流式的 AsyncResult：

>>> import asyncio

>>> from sqlalchemy import Column
>>> from sqlalchemy import MetaData
>>> from sqlalchemy import select
>>> from sqlalchemy import String
>>> from sqlalchemy import Table
>>> from sqlalchemy.ext.asyncio import create_async_engine

>>> meta = MetaData()
>>> t1 = Table("t1", meta, Column("name", String(50), primary_key=True))


>>> async def async_main() -> None:
...     engine = create_async_engine("sqlite+aiosqlite://", echo=True)
...
...     async with engine.begin() as conn:
...         await conn.run_sync(meta.drop_all)
...         await conn.run_sync(meta.create_all)
...
...         await conn.execute(
...             t1.insert(), [{"name": "some name 1"}, {"name": "some name 2"}]
...         )
...
...     async with engine.connect() as conn:
...         # 执行 select 返回一个 Result，将以缓冲方式返回结果
...         result = await conn.execute(select(t1).where(t1.c.name == "some name 1"))
...
...         print(result.fetchall())
...
...     # 对于在函数作用域中创建的 AsyncEngine，需显式关闭并清理连接池中的连接
...     await engine.dispose()


>>> asyncio.run(async_main())
BEGIN (implicit)
...
CREATE TABLE t1 (
    name VARCHAR(50) NOT NULL,
    PRIMARY KEY (name)
)
...
INSERT INTO t1 (name) VALUES (?)
[...] [('some name 1',), ('some name 2',)]
COMMIT
BEGIN (implicit)
SELECT t1.name
FROM t1
WHERE t1.name = ?
[...] ('some name 1',)
[('some name 1',)]
ROLLBACK

如上所示，可以使用 AsyncConnection.run_sync() 方法调用特殊的 DDL 函数， 如 MetaData.create_all()，这类函数没有可等待（awaitable）钩子。

小技巧

当在一个即将离开上下文并被垃圾回收的作用域中使用 AsyncEngine 对象时， 建议使用 await 调用 AsyncEngine.dispose() 方法，如上述示例中的 async_main 函数所示。 这样可以确保连接池中所有打开的连接能在可等待上下文中正确关闭。 与阻塞式 IO 不同，SQLAlchemy 无法在诸如 __del__ 或 weakref finalizer 这类方法中正确释放这些连接， 因为这些地方无法使用 await。如果未显式处理引擎的释放，当其超出作用域时，可能会在垃圾回收过程中产生类似 RuntimeError: Event loop is closed 的警告信息。

AsyncConnection 还通过 AsyncConnection.stream() 方法提供了 “流式” API，返回一个 AsyncResult 对象。该结果对象使用服务器端游标，并提供 async/await API， 例如异步迭代器:

async with engine.connect() as conn:
    async_result = await conn.stream(select(t1))

    async for row in async_result:
        print("row: %s" % (row,))

英文

For Core use, the create_async_engine() function creates an instance of AsyncEngine which then offers an async version of the traditional Engine API. The AsyncEngine delivers an AsyncConnection via its AsyncEngine.connect() and AsyncEngine.begin() methods which both deliver asynchronous context managers. The AsyncConnection can then invoke statements using either the AsyncConnection.execute() method to deliver a buffered Result, or the AsyncConnection.stream() method to deliver a streaming server-side AsyncResult:

>>> import asyncio

>>> from sqlalchemy import Column
>>> from sqlalchemy import MetaData
>>> from sqlalchemy import select
>>> from sqlalchemy import String
>>> from sqlalchemy import Table
>>> from sqlalchemy.ext.asyncio import create_async_engine

>>> meta = MetaData()
>>> t1 = Table("t1", meta, Column("name", String(50), primary_key=True))


>>> async def async_main() -> None:
...     engine = create_async_engine("sqlite+aiosqlite://", echo=True)
...
...     async with engine.begin() as conn:
...         await conn.run_sync(meta.drop_all)
...         await conn.run_sync(meta.create_all)
...
...         await conn.execute(
...             t1.insert(), [{"name": "some name 1"}, {"name": "some name 2"}]
...         )
...
...     async with engine.connect() as conn:
...         # select a Result, which will be delivered with buffered
...         # results
...         result = await conn.execute(select(t1).where(t1.c.name == "some name 1"))
...
...         print(result.fetchall())
...
...     # for AsyncEngine created in function scope, close and
...     # clean-up pooled connections
...     await engine.dispose()


>>> asyncio.run(async_main())
BEGIN (implicit)
...
CREATE TABLE t1 (
    name VARCHAR(50) NOT NULL,
    PRIMARY KEY (name)
)
...
INSERT INTO t1 (name) VALUES (?)
[...] [('some name 1',), ('some name 2',)]
COMMIT
BEGIN (implicit)
SELECT t1.name
FROM t1
WHERE t1.name = ?
[...] ('some name 1',)
[('some name 1',)]
ROLLBACK

Above, the AsyncConnection.run_sync() method may be used to invoke special DDL functions such as MetaData.create_all() that don’t include an awaitable hook.

小技巧

It’s advisable to invoke the AsyncEngine.dispose() method

using await when using the AsyncEngine object in a scope that will go out of context and be garbage collected, as illustrated in the async_main function in the above example. This ensures that any connections held open by the connection pool will be properly disposed within an awaitable context. Unlike when using blocking IO, SQLAlchemy cannot properly dispose of these connections within methods like __del__ or weakref finalizers as there is no opportunity to invoke await. Failing to explicitly dispose of the engine when it falls out of scope may result in warnings emitted to standard out resembling the form RuntimeError: Event loop is closed within garbage collection.

The AsyncConnection also features a “streaming” API via the AsyncConnection.stream() method that returns an AsyncResult object. This result object uses a server-side cursor and provides an async/await API, such as an async iterator:

async with engine.connect() as conn:
    async_result = await conn.stream(select(t1))

    async for row in async_result:
        print("row: %s" % (row,))

概要 - ORM

Synopsis - ORM

中文

使用 2.0 style 查询风格时，AsyncSession 类提供完整的 ORM 功能。

在默认的使用模式下，需特别注意避免 lazy loading 或其他涉及 ORM 关系和列属性的 过期属性访问；下一节 使用 AsyncSession 时防止隐式 IO 将对此进行详细说明。

警告

单个 AsyncSession 实例在 多个并发任务中使用并不安全。 详见 将 AsyncSession 与并发任务结合使用 和 会话是线程安全的吗？AsyncSession 在并发任务中共享是否安全？ 章节了解背景信息。

以下示例展示了一个完整的例子，包括映射器和会话的配置：

>>> from __future__ import annotations

>>> import asyncio
>>> import datetime
>>> from typing import List

>>> from sqlalchemy import ForeignKey
>>> from sqlalchemy import func
>>> from sqlalchemy import select
>>> from sqlalchemy.ext.asyncio import AsyncAttrs
>>> from sqlalchemy.ext.asyncio import async_sessionmaker
>>> from sqlalchemy.ext.asyncio import AsyncSession
>>> from sqlalchemy.ext.asyncio import create_async_engine
>>> from sqlalchemy.orm import DeclarativeBase
>>> from sqlalchemy.orm import Mapped
>>> from sqlalchemy.orm import mapped_column
>>> from sqlalchemy.orm import relationship
>>> from sqlalchemy.orm import selectinload


>>> class Base(AsyncAttrs, DeclarativeBase):
...     pass

>>> class B(Base):
...     __tablename__ = "b"
...
...     id: Mapped[int] = mapped_column(primary_key=True)
...     a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))
...     data: Mapped[str]

>>> class A(Base):
...     __tablename__ = "a"
...
...     id: Mapped[int] = mapped_column(primary_key=True)
...     data: Mapped[str]
...     create_date: Mapped[datetime.datetime] = mapped_column(server_default=func.now())
...     bs: Mapped[List[B]] = relationship()

>>> async def insert_objects(async_session: async_sessionmaker[AsyncSession]) -> None:
...     async with async_session() as session:
...         async with session.begin():
...             session.add_all(
...                 [
...                     A(bs=[B(data="b1"), B(data="b2")], data="a1"),
...                     A(bs=[], data="a2"),
...                     A(bs=[B(data="b3"), B(data="b4")], data="a3"),
...                 ]
...             )


>>> async def select_and_update_objects(
...     async_session: async_sessionmaker[AsyncSession],
... ) -> None:
...     async with async_session() as session:
...         stmt = select(A).order_by(A.id).options(selectinload(A.bs))
...
...         result = await session.execute(stmt)
...
...         for a in result.scalars():
...             print(a, a.data)
...             print(f"created at: {a.create_date}")
...             for b in a.bs:
...                 print(b, b.data)
...
...         result = await session.execute(select(A).order_by(A.id).limit(1))
...
...         a1 = result.scalars().one()
...
...         a1.data = "new data"
...
...         await session.commit()
...
...         # 提交后访问属性；这正是 expire_on_commit=False 所允许的
...         print(a1.data)
...
...         # 或者，AsyncAttrs 允许将任何属性作为 awaitable 来访问（2.0.13 中新增）
...         for b1 in await a1.awaitable_attrs.bs:
...             print(b1, b1.data)


>>> async def async_main() -> None:
...     engine = create_async_engine("sqlite+aiosqlite://", echo=True)
...
...     # async_sessionmaker：用于创建 AsyncSession 对象的工厂。
...     # 设置 expire_on_commit=False，避免事务提交后使对象失效
...     async_session = async_sessionmaker(engine, expire_on_commit=False)
...
...     async with engine.begin() as conn:
...         await conn.run_sync(Base.metadata.create_all)
...
...     await insert_objects(async_session)
...     await select_and_update_objects(async_session)
...
...     # 对于在函数作用域中创建的 AsyncEngine，需显式关闭并清理连接池连接
...     await engine.dispose()


>>> asyncio.run(async_main())
BEGIN (implicit)
...
CREATE TABLE a (
    id INTEGER NOT NULL,
    data VARCHAR NOT NULL,
    create_date DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
    PRIMARY KEY (id)
)
...
CREATE TABLE b (
    id INTEGER NOT NULL,
    a_id INTEGER NOT NULL,
    data VARCHAR NOT NULL,
    PRIMARY KEY (id),
    FOREIGN KEY(a_id) REFERENCES a (id)
)
...
COMMIT
BEGIN (implicit)
INSERT INTO a (data) VALUES (?) RETURNING id, create_date
[...] ('a1',)
...
INSERT INTO b (a_id, data) VALUES (?, ?) RETURNING id
[...] (1, 'b2')
...
COMMIT
BEGIN (implicit)
SELECT a.id, a.data, a.create_date
FROM a ORDER BY a.id
[...] ()
SELECT b.a_id AS b_a_id, b.id AS b_id, b.data AS b_data
FROM b
WHERE b.a_id IN (?, ?, ?)
[...] (1, 2, 3)
<A object at ...> a1
created at: ...
<B object at ...> b1
<B object at ...> b2
<A object at ...> a2
created at: ...
<A object at ...> a3
created at: ...
<B object at ...> b3
<B object at ...> b4
SELECT a.id, a.data, a.create_date
FROM a ORDER BY a.id
LIMIT ? OFFSET ?
[...] (1, 0)
UPDATE a SET data=? WHERE a.id = ?
[...] ('new data', 1)
COMMIT
new data
<B object at ...> b1
<B object at ...> b2

在上述示例中，AsyncSession 是通过可选的辅助工具 async_sessionmaker 实例化的，它提供了一个创建新的 AsyncSession 对象的工厂，具备一组固定参数， 其中包括将其绑定到某个特定数据库 URL 的 AsyncEngine。 之后，它被传递给其他方法，并在 Python 异步上下文管理器（即 async with: 语句）中使用， 以便在代码块结束时自动关闭；其行为等同于调用 AsyncSession.close() 方法。

英文

Using 2.0 style querying, the AsyncSession class provides full ORM functionality.

Within the default mode of use, special care must be taken to avoid lazy loading or other expired-attribute access involving ORM relationships and column attributes; the next section 使用 AsyncSession 时防止隐式 IO details this.

警告

A single instance of AsyncSession is not safe for use in multiple, concurrent tasks. See the sections 将 AsyncSession 与并发任务结合使用 and 会话是线程安全的吗？AsyncSession 在并发任务中共享是否安全？ for background.

The example below illustrates a complete example including mapper and session configuration:

>>> from __future__ import annotations

>>> import asyncio
>>> import datetime
>>> from typing import List

>>> from sqlalchemy import ForeignKey
>>> from sqlalchemy import func
>>> from sqlalchemy import select
>>> from sqlalchemy.ext.asyncio import AsyncAttrs
>>> from sqlalchemy.ext.asyncio import async_sessionmaker
>>> from sqlalchemy.ext.asyncio import AsyncSession
>>> from sqlalchemy.ext.asyncio import create_async_engine
>>> from sqlalchemy.orm import DeclarativeBase
>>> from sqlalchemy.orm import Mapped
>>> from sqlalchemy.orm import mapped_column
>>> from sqlalchemy.orm import relationship
>>> from sqlalchemy.orm import selectinload


>>> class Base(AsyncAttrs, DeclarativeBase):
...     pass

>>> class B(Base):
...     __tablename__ = "b"
...
...     id: Mapped[int] = mapped_column(primary_key=True)
...     a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))
...     data: Mapped[str]

>>> class A(Base):
...     __tablename__ = "a"
...
...     id: Mapped[int] = mapped_column(primary_key=True)
...     data: Mapped[str]
...     create_date: Mapped[datetime.datetime] = mapped_column(server_default=func.now())
...     bs: Mapped[List[B]] = relationship()

>>> async def insert_objects(async_session: async_sessionmaker[AsyncSession]) -> None:
...     async with async_session() as session:
...         async with session.begin():
...             session.add_all(
...                 [
...                     A(bs=[B(data="b1"), B(data="b2")], data="a1"),
...                     A(bs=[], data="a2"),
...                     A(bs=[B(data="b3"), B(data="b4")], data="a3"),
...                 ]
...             )


>>> async def select_and_update_objects(
...     async_session: async_sessionmaker[AsyncSession],
... ) -> None:
...     async with async_session() as session:
...         stmt = select(A).order_by(A.id).options(selectinload(A.bs))
...
...         result = await session.execute(stmt)
...
...         for a in result.scalars():
...             print(a, a.data)
...             print(f"created at: {a.create_date}")
...             for b in a.bs:
...                 print(b, b.data)
...
...         result = await session.execute(select(A).order_by(A.id).limit(1))
...
...         a1 = result.scalars().one()
...
...         a1.data = "new data"
...
...         await session.commit()
...
...         # access attribute subsequent to commit; this is what
...         # expire_on_commit=False allows
...         print(a1.data)
...
...         # alternatively, AsyncAttrs may be used to access any attribute
...         # as an awaitable (new in 2.0.13)
...         for b1 in await a1.awaitable_attrs.bs:
...             print(b1, b1.data)


>>> async def async_main() -> None:
...     engine = create_async_engine("sqlite+aiosqlite://", echo=True)
...
...     # async_sessionmaker: a factory for new AsyncSession objects.
...     # expire_on_commit - don't expire objects after transaction commit
...     async_session = async_sessionmaker(engine, expire_on_commit=False)
...
...     async with engine.begin() as conn:
...         await conn.run_sync(Base.metadata.create_all)
...
...     await insert_objects(async_session)
...     await select_and_update_objects(async_session)
...
...     # for AsyncEngine created in function scope, close and
...     # clean-up pooled connections
...     await engine.dispose()


>>> asyncio.run(async_main())
BEGIN (implicit)
...
CREATE TABLE a (
    id INTEGER NOT NULL,
    data VARCHAR NOT NULL,
    create_date DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
    PRIMARY KEY (id)
)
...
CREATE TABLE b (
    id INTEGER NOT NULL,
    a_id INTEGER NOT NULL,
    data VARCHAR NOT NULL,
    PRIMARY KEY (id),
    FOREIGN KEY(a_id) REFERENCES a (id)
)
...
COMMIT
BEGIN (implicit)
INSERT INTO a (data) VALUES (?) RETURNING id, create_date
[...] ('a1',)
...
INSERT INTO b (a_id, data) VALUES (?, ?) RETURNING id
[...] (1, 'b2')
...
COMMIT
BEGIN (implicit)
SELECT a.id, a.data, a.create_date
FROM a ORDER BY a.id
[...] ()
SELECT b.a_id AS b_a_id, b.id AS b_id, b.data AS b_data
FROM b
WHERE b.a_id IN (?, ?, ?)
[...] (1, 2, 3)
<A object at ...> a1
created at: ...
<B object at ...> b1
<B object at ...> b2
<A object at ...> a2
created at: ...
<A object at ...> a3
created at: ...
<B object at ...> b3
<B object at ...> b4
SELECT a.id, a.data, a.create_date
FROM a ORDER BY a.id
LIMIT ? OFFSET ?
[...] (1, 0)
UPDATE a SET data=? WHERE a.id = ?
[...] ('new data', 1)
COMMIT
new data
<B object at ...> b1
<B object at ...> b2

In the example above, the AsyncSession is instantiated using the optional async_sessionmaker helper, which provides a factory for new AsyncSession objects with a fixed set of parameters, which here includes associating it with an AsyncEngine against particular database URL. It is then passed to other methods where it may be used in a Python asynchronous context manager (i.e. async with: statement) so that it is automatically closed at the end of the block; this is equivalent to calling the AsyncSession.close() method.

将 AsyncSession 与并发任务结合使用

Using AsyncSession with Concurrent Tasks

中文

AsyncSession 对象是一个 可变的、有状态的对象 ，表示 单个有状态的数据库事务正在进行 。使用 asyncio 的并发任务时，例如使用 asyncio.gather() 之类的 API，每个单独任务应该使用一个 单独的 AsyncSession。

有关 Session 和 AsyncSession 在并发工作负载中的使用方式的一般说明，请参阅 会话是线程安全的吗？AsyncSession 在并发任务中共享是否安全？ 部分。

英文

The AsyncSession object is a mutable, stateful object which represents a single, stateful database transaction in progress. Using concurrent tasks with asyncio, with APIs such as asyncio.gather() for example, should use a separate AsyncSession per individual task.

See the section 会话是线程安全的吗？AsyncSession 在并发任务中共享是否安全？ for a general description of the Session and AsyncSession with regards to how they should be used with concurrent workloads.

使用 AsyncSession 时防止隐式 IO

Preventing Implicit IO when Using AsyncSession

以下是上述内容的翻译：

—

使用传统的 asyncio 时，应用程序需要避免在属性访问时隐式触发 IO 的情况。以下是一些可行的技术手段，许多已在前文示例中展示。

• 对于延迟加载关系、延迟列或表达式，以及在属性过期的情况下访问属性的场景，可以使用 AsyncAttrs 混入类。将此混入类添加到具体类，或更通用地添加到声明式 Base 超类中，即可通过 AsyncAttrs.awaitable_attrs 访问器将任何属性作为可等待对象访问:

  from __future__ import annotations
  from typing import List
  from sqlalchemy.ext.asyncio import AsyncAttrs
  from sqlalchemy.orm import DeclarativeBase, Mapped, relationship
  
  class Base(AsyncAttrs, DeclarativeBase):
      pass
  
  class A(Base):
      __tablename__ = "a"
      bs: Mapped[List[B]] = relationship()
  
  class B(Base):
      __tablename__ = "b"

当未使用预加载时，访问新加载实例的 A.bs 集合通常会触发 lazy loading，这通常需要执行数据库 IO。在 asyncio 中不允许隐式 IO，因此可以使用 awaitable_attrs 前缀以可等待方式访问该属性:

a1 = (await session.scalars(select(A))).one()
for b1 in await a1.awaitable_attrs.bs:
    print(b1)

AsyncAttrs 提供了一个简洁的接口，其底层逻辑也被 AsyncSession.run_sync() 方法所使用。

在 2.0.13 版本加入.

• 可以使用 仅写集合（write only collections） 替代集合属性，这样将永远不会隐式执行 IO。此功能可参考 SQLAlchemy 2.0 中的 只写关系。使用该功能时，集合不会被读取，只能通过显式 SQL 查询。可参考 Asyncio 集成 中的 async_orm_writeonly.py 示例。

使用仅写集合时，程序行为简单、可预测。但缺点是无法批量加载集合，必须手动处理。因此以下要点仍涉及常规的懒加载关系的处理技巧。

• 如果不使用 AsyncAttrs，可以将关系声明为 lazy="raise"，避免其尝试执行 SQL。此时应改为使用 eager loading。

• 最实用的预加载策略是 selectinload()，如前面的例子所示，用于预加载 A.bs 集合:

  stmt = select(A).options(selectinload(A.bs))

• 创建新对象时， 集合应总是赋默认值，例如空列表:

  A(bs=[], data="a2")

这使得在对象刷新前 .bs 属性即可读取。否则在刷新时该集合属性会处于未加载状态，访问时将报错。

• AsyncSession 应设置 Session.expire_on_commit=False，以便在调用 AsyncSession.commit() 后仍可访问对象属性。例如:

  async_session = async_sessionmaker(engine, expire_on_commit=False)
  
  async with async_session() as session:
      result = await session.execute(select(A).order_by(A.id))
      a1 = result.scalars().first()
      await session.commit()
      print(a1.data)  # commit 后仍可访问

其他注意事项包括：

• 应避免使用 AsyncSession.expire()，若确实需要可使用 AsyncSession.refresh()。一般不建议使用过期机制，通常应将 expire_on_commit 设为 False。

• 在 asyncio 下可以显式加载懒加载关系，方法是将属性名传递给 AsyncSession.refresh() 的 attribute_names 参数:

  a_obj = await async_session.get(A, [1])
  await async_session.refresh(a_obj, ["bs"])
  print(f"bs collection: {a_obj.bs}")

在 2.0.4 版本加入: 支持通过 refresh 的 attribute_names 显式加载关系属性。

• 避免使用 all 级联选项（详见 级联），应明确列出所需的级联功能。 all 包含 刷新-过期，它会使 AsyncSession.refresh() 使相关对象属性过期，却不一定重新加载这些对象。

• 对于 deferred() 列，应使用合适的加载选项，方式与 relationship() 类似。详见 使用【列延迟】限制加载哪些列。

• 默认情况下，“动态”关系加载器（详见 动态关系加载器）不兼容 asyncio。可以通过 AsyncSession.run_sync() 使用，或使用其 .statement 属性构造普通 select 查询:

  user = await session.get(User, 42)
  addresses = (await session.scalars(user.addresses.statement)).all()
  
  stmt = user.addresses.statement.where(Address.email_address.startswith("patrick"))
  addresses_filter = (await session.scalars(stmt)).all()

SQLAlchemy 2.0 引入的 write_only 技术完全兼容 asyncio，建议优先使用。

参见

“Dynamic” relationship loaders superseded by “Write Only” - 迁移到 2.0 风格的注意事项

• 如果在 asyncio 中使用的数据库（如 MySQL 8）不支持 RETURNING 子句，那么服务器默认值（如自动生成的时间戳）将无法在对象刷新后获得。除非启用 Mapper.eager_defaults。在 SQLAlchemy 2.0 中，对于支持 RETURNING 的数据库（如 PostgreSQL、SQLite 和 MariaDB）会自动启用此行为。

英文

Using traditional asyncio, the application needs to avoid any points at which IO-on-attribute access may occur. Techniques that can be used to help this are below, many of which are illustrated in the preceding example.

• Attributes that are lazy-loading relationships, deferred columns or expressions, or are being accessed in expiration scenarios can take advantage of the AsyncAttrs mixin. This mixin, when added to a specific class or more generally to the Declarative Base superclass, provides an accessor AsyncAttrs.awaitable_attrs which delivers any attribute as an awaitable:

  from __future__ import annotations
  
  from typing import List
  
  from sqlalchemy.ext.asyncio import AsyncAttrs
  from sqlalchemy.orm import DeclarativeBase
  from sqlalchemy.orm import Mapped
  from sqlalchemy.orm import relationship
  
  
  class Base(AsyncAttrs, DeclarativeBase):
      pass
  
  
  class A(Base):
      __tablename__ = "a"
  
      # ... rest of mapping ...
  
      bs: Mapped[List[B]] = relationship()
  
  
  class B(Base):
      __tablename__ = "b"
  
      # ... rest of mapping ...

  Accessing the A.bs collection on newly loaded instances of A when eager loading is not in use will normally use lazy loading, which in order to succeed will usually emit IO to the database, which will fail under asyncio as no implicit IO is allowed. To access this attribute directly under asyncio without any prior loading operations, the attribute can be accessed as an awaitable by indicating the AsyncAttrs.awaitable_attrs prefix:

  a1 = (await session.scalars(select(A))).one()
  for b1 in await a1.awaitable_attrs.bs:
      print(b1)

  The AsyncAttrs mixin provides a succinct facade over the internal approach that’s also used by the AsyncSession.run_sync() method.

  在 2.0.13 版本加入.

  参见

  AsyncAttrs

• Collections can be replaced with write only collections that will never emit IO implicitly, by using the 只写关系 feature in SQLAlchemy 2.0. Using this feature, collections are never read from, only queried using explicit SQL calls. See the example async_orm_writeonly.py in the Asyncio 集成 section for an example of write-only collections used with asyncio.

  When using write only collections, the program’s behavior is simple and easy to predict regarding collections. However, the downside is that there is not any built-in system for loading many of these collections all at once, which instead would need to be performed manually. Therefore, many of the bullets below address specific techniques when using traditional lazy-loaded relationships with asyncio, which requires more care.

• If not using AsyncAttrs, relationships can be declared with lazy="raise" so that by default they will not attempt to emit SQL. In order to load collections, eager loading would be used instead.

• The most useful eager loading strategy is the selectinload() eager loader, which is employed in the previous example in order to eagerly load the A.bs collection within the scope of the await session.execute() call:

  stmt = select(A).options(selectinload(A.bs))

• When constructing new objects, collections are always assigned a default, empty collection, such as a list in the above example:

  A(bs=[], data="a2")

  This allows the .bs collection on the above A object to be present and readable when the A object is flushed; otherwise, when the A is flushed, .bs would be unloaded and would raise an error on access.

• The AsyncSession is configured using Session.expire_on_commit set to False, so that we may access attributes on an object subsequent to a call to AsyncSession.commit(), as in the line at the end where we access an attribute:

  # create AsyncSession with expire_on_commit=False
  async_session = AsyncSession(engine, expire_on_commit=False)
  
  # sessionmaker version
  async_session = async_sessionmaker(engine, expire_on_commit=False)
  
  async with async_session() as session:
      result = await session.execute(select(A).order_by(A.id))
  
      a1 = result.scalars().first()
  
      # commit would normally expire all attributes
      await session.commit()
  
      # access attribute subsequent to commit; this is what
      # expire_on_commit=False allows
      print(a1.data)

Other guidelines include:

• Methods like AsyncSession.expire() should be avoided in favor of AsyncSession.refresh(); if expiration is absolutely needed. Expiration should generally not be needed as Session.expire_on_commit should normally be set to False when using asyncio.

• A lazy-loaded relationship can be loaded explicitly under asyncio using AsyncSession.refresh(), if the desired attribute name is passed explicitly to Session.refresh.attribute_names, e.g.:

  # assume a_obj is an A that has lazy loaded A.bs collection
  a_obj = await async_session.get(A, [1])
  
  # force the collection to load by naming it in attribute_names
  await async_session.refresh(a_obj, ["bs"])
  
  # collection is present
  print(f"bs collection: {a_obj.bs}")

  It’s of course preferable to use eager loading up front in order to have collections already set up without the need to lazy-load.

  在 2.0.4 版本加入: Added support for AsyncSession.refresh() and the underlying Session.refresh() method to force lazy-loaded relationships to load, if they are named explicitly in the Session.refresh.attribute_names parameter. In previous versions, the relationship would be silently skipped even if named in the parameter.

• Avoid using the all cascade option documented at 级联 in favor of listing out the desired cascade features explicitly. The all cascade option implies among others the 刷新-过期 setting, which means that the AsyncSession.refresh() method will expire the attributes on related objects, but not necessarily refresh those related objects assuming eager loading is not configured within the relationship(), leaving them in an expired state.

• Appropriate loader options should be employed for deferred() columns, if used at all, in addition to that of relationship() constructs as noted above. See 使用【列延迟】限制加载哪些列 for background on deferred column loading.

• The “dynamic” relationship loader strategy described at 动态关系加载器 is not compatible by default with the asyncio approach. It can be used directly only if invoked within the AsyncSession.run_sync() method described at 在 asyncio 下运行同步方法和函数, or by using its .statement attribute to obtain a normal select:

  user = await session.get(User, 42)
  addresses = (await session.scalars(user.addresses.statement)).all()
  stmt = user.addresses.statement.where(Address.email_address.startswith("patrick"))
  addresses_filter = (await session.scalars(stmt)).all()

  The write only technique, introduced in version 2.0 of SQLAlchemy, is fully compatible with asyncio and should be preferred.

  参见

  “Dynamic” relationship loaders superseded by “Write Only” - notes on migration to 2.0 style

• If using asyncio with a database that does not support RETURNING, such as MySQL 8, server default values such as generated timestamps will not be available on newly flushed objects unless the Mapper.eager_defaults option is used. In SQLAlchemy 2.0, this behavior is applied automatically to backends like PostgreSQL, SQLite and MariaDB which use RETURNING to fetch new values when rows are INSERTed.

在 asyncio 下运行同步方法和函数

Running Synchronous Methods and Functions under asyncio

中文

Deep Alchemy

这种方式本质上是公开了 SQLAlchemy 用以提供 asyncio 接口的底层机制。 虽然从技术角度看这样做并没有问题，但总体上可以认为该方式较为“有争议”， 因为它违背了 asyncio 编程模型的一些核心理念。asyncio 的基本理念是： 任何可能触发 IO 的编程语句 必须 通过 await 来显式标识，否则程序就无法明确指出在哪一行可能会发生 IO。 这种方法并未改变该理念的本质，只是允许在函数调用范围内，将一系列同步的 IO 指令打包为一个 awaitable， 从而免除该规则的限制。

作为将传统 SQLAlchemy “懒加载” 集成到 asyncio 事件循环中的一种 可选 方法， SQLAlchemy 提供了 AsyncSession.run_sync()，它可以在 greenlet 中运行任意 Python 函数， 并在访问数据库驱动时将传统同步编程逻辑自动转换为 await。一个典型做法是， asyncio 风格的应用可以将数据库相关的逻辑封装为函数，并通过 AsyncSession.run_sync() 来调用它们。

改写上面的示例，如果我们没有使用 selectinload() 来预加载 A.bs 集合， 可以将相关的属性访问操作封装在一个单独函数中执行:

import asyncio

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine


def fetch_and_update_objects(session):
    """在一个 awaitable 函数中运行传统同步风格的 ORM 代码。"""

    # 此处的 session 是传统 ORM Session。
    # 所有功能都可用，包括旧版 Query 接口。

    stmt = select(A)

    result = session.execute(stmt)
    for a1 in result.scalars():
        print(a1)

        # 懒加载
        for b1 in a1.bs:
            print(b1)

    # 使用 legacy Query 接口
    a1 = session.query(A).order_by(A.id).first()

    a1.data = "new data"


async def async_main():
    engine = create_async_engine(
        "postgresql+asyncpg://scott:tiger@localhost/test",
        echo=True,
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    async with AsyncSession(engine) as session:
        async with session.begin():
            session.add_all(
                [
                    A(bs=[B(), B()], data="a1"),
                    A(bs=[B()], data="a2"),
                    A(bs=[B(), B()], data="a3"),
                ]
            )

        await session.run_sync(fetch_and_update_objects)

        await session.commit()

    # 如果在函数作用域中创建了 AsyncEngine，关闭并清理连接池
    await engine.dispose()


asyncio.run(async_main())

以上方式通过在 “同步” 运行器中执行某些函数，在某种程度上类似于在基于事件的编程库（如 gevent）之上运行 SQLAlchemy 应用程序。 不同之处如下：

1. 与使用 gevent 不同，我们可以继续使用标准的 Python asyncio 事件循环或任何自定义事件循环，无需集成到 gevent 的事件循环中。

2. 完全不涉及 “monkeypatching”。上述示例使用的是原生 asyncio 驱动，底层 SQLAlchemy 连接池也使用 Python 内置的 asyncio.Queue 进行连接池管理。

3. 程序可以在 async/await 异步代码与包含同步代码的函数之间自由切换，几乎不会产生性能开销。 这不依赖于线程执行器（thread executor），也没有额外的等待器或同步机制。

4. 底层网络驱动程序完全使用 Python 的 asyncio 概念， 不使用任何第三方网络库（如 gevent 和 eventlet 所提供的网络功能）。

英文

Deep Alchemy

This approach is essentially exposing publicly the mechanism by which SQLAlchemy is able to provide the asyncio interface in the first place. While there is no technical issue with doing so, overall the approach can probably be considered “controversial” as it works against some of the central philosophies of the asyncio programming model, which is essentially that any programming statement that can potentially result in IO being invoked must have an await call, lest the program does not make it explicitly clear every line at which IO may occur. This approach does not change that general idea, except that it allows a series of synchronous IO instructions to be exempted from this rule within the scope of a function call, essentially bundled up into a single awaitable.

As an alternative means of integrating traditional SQLAlchemy “lazy loading” within an asyncio event loop, an optional method known as AsyncSession.run_sync() is provided which will run any Python function inside of a greenlet, where traditional synchronous programming concepts will be translated to use await when they reach the database driver. A hypothetical approach here is an asyncio-oriented application can package up database-related methods into functions that are invoked using AsyncSession.run_sync().

Altering the above example, if we didn’t use selectinload() for the A.bs collection, we could accomplish our treatment of these attribute accesses within a separate function:

import asyncio

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine


def fetch_and_update_objects(session):
    """run traditional sync-style ORM code in a function that will be
    invoked within an awaitable.

    """

    # the session object here is a traditional ORM Session.
    # all features are available here including legacy Query use.

    stmt = select(A)

    result = session.execute(stmt)
    for a1 in result.scalars():
        print(a1)

        # lazy loads
        for b1 in a1.bs:
            print(b1)

    # legacy Query use
    a1 = session.query(A).order_by(A.id).first()

    a1.data = "new data"


async def async_main():
    engine = create_async_engine(
        "postgresql+asyncpg://scott:tiger@localhost/test",
        echo=True,
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    async with AsyncSession(engine) as session:
        async with session.begin():
            session.add_all(
                [
                    A(bs=[B(), B()], data="a1"),
                    A(bs=[B()], data="a2"),
                    A(bs=[B(), B()], data="a3"),
                ]
            )

        await session.run_sync(fetch_and_update_objects)

        await session.commit()

    # for AsyncEngine created in function scope, close and
    # clean-up pooled connections
    await engine.dispose()


asyncio.run(async_main())

The above approach of running certain functions within a “sync” runner has some parallels to an application that runs a SQLAlchemy application on top of an event-based programming library such as gevent. The differences are as follows:

1. unlike when using gevent, we can continue to use the standard Python asyncio event loop, or any custom event loop, without the need to integrate into the gevent event loop.

2. There is no “monkeypatching” whatsoever. The above example makes use of a real asyncio driver and the underlying SQLAlchemy connection pool is also using the Python built-in asyncio.Queue for pooling connections.

3. The program can freely switch between async/await code and contained functions that use sync code with virtually no performance penalty. There is no “thread executor” or any additional waiters or synchronization in use.

4. The underlying network drivers are also using pure Python asyncio concepts, no third party networking libraries as gevent and eventlet provides are in use.

使用 asyncio 扩展的事件

Using events with the asyncio extension

中文

SQLAlchemy 的 事件系统 并未被 asyncio 扩展直接暴露， 这意味着当前还没有 “async” 版本的 SQLAlchemy 事件处理器。

不过，由于 asyncio 扩展是围绕传统同步 SQLAlchemy API 构建的， 因此仍然可以自由使用常规的 “同步” 风格事件处理器，就像不使用 asyncio 时一样。

如下面详细说明的，当前有两种策略可以在 asyncio 接口中注册事件：

• 可以在实例级别注册事件（例如，某个特定的 AsyncEngine 实例）， 方法是将事件绑定到该实例对应的 sync 属性上，该属性指向被代理的同步对象。 例如，若要在某个 AsyncEngine 实例上注册 PoolEvents.connect() 事件，应使用其 AsyncEngine.sync_engine 属性作为目标。可用的目标包括：

  AsyncEngine.sync_engine

  AsyncConnection.sync_connection

  AsyncConnection.sync_engine

  AsyncSession.sync_session

• 若要在类级别注册事件（即目标是所有该类型的实例，例如所有 AsyncSession 实例），则应使用对应的同步风格类作为目标。 例如，若要为 AsyncSession 类注册 SessionEvents.before_commit() 事件，应使用 Session 类作为目标。

• 若要在 sessionmaker 层级注册事件，可结合使用显式声明的 sessionmaker 与 async_sessionmaker， 通过 async_sessionmaker.sync_session_class 参数进行关联， 并将事件绑定到 sessionmaker 上。

在 asyncio 上下文中使用事件处理器时，诸如 Connection 之类的对象 仍然以其传统的 “同步” 方式工作，无需使用 await 或 async； 当消息最终传递到 asyncio 数据库适配器时，调用方式将被透明地转换为 asyncio 风格的调用。 对于接收 DBAPI 层连接对象的事件（如 PoolEvents.connect()）， 所传入的对象是符合 pep-249 标准的 “connection” 对象， 它将同步风格的调用适配为 asyncio 驱动调用。

英文

The SQLAlchemy event system is not directly exposed by the asyncio extension, meaning there is not yet an “async” version of a SQLAlchemy event handler.

However, as the asyncio extension surrounds the usual synchronous SQLAlchemy API, regular “synchronous” style event handlers are freely available as they would be if asyncio were not used.

As detailed below, there are two current strategies to register events given asyncio-facing APIs:

• Events can be registered at the instance level (e.g. a specific AsyncEngine instance) by associating the event with the sync attribute that refers to the proxied object. For example to register the PoolEvents.connect() event against an AsyncEngine instance, use its AsyncEngine.sync_engine attribute as target. Targets include:

  AsyncEngine.sync_engine

  AsyncConnection.sync_connection

  AsyncConnection.sync_engine

  AsyncSession.sync_session

• To register an event at the class level, targeting all instances of the same type (e.g. all AsyncSession instances), use the corresponding sync-style class. For example to register the SessionEvents.before_commit() event against the AsyncSession class, use the Session class as the target.

• To register at the sessionmaker level, combine an explicit sessionmaker with an async_sessionmaker using async_sessionmaker.sync_session_class, and associate events with the sessionmaker.

When working within an event handler that is within an asyncio context, objects like the Connection continue to work in their usual “synchronous” way without requiring await or async usage; when messages are ultimately received by the asyncio database adapter, the calling style is transparently adapted back into the asyncio calling style. For events that are passed a DBAPI level connection, such as PoolEvents.connect(), the object is a pep-249 compliant “connection” object which will adapt sync-style calls into the asyncio driver.

异步引擎/会话/会话生成器的事件监听器示例

Examples of Event Listeners with Async Engines / Sessions / Sessionmakers

中文

下面是一些将同步风格事件处理器与面向异步的 API 构造关联的示例：

• AsyncEngine 上的 Core 事件

  在此示例中，我们通过 AsyncEngine 的 AsyncEngine.sync_engine 属性，作为 ConnectionEvents 和 PoolEvents 的目标来访问同步引擎实例:

  import asyncio
  
  from sqlalchemy import event
  from sqlalchemy import text
  from sqlalchemy.engine import Engine
  from sqlalchemy.ext.asyncio import create_async_engine
  
  engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost:5432/test")
  
  
  # 在 Engine 实例上注册 connect 事件
  @event.listens_for(engine.sync_engine, "connect")
  def my_on_connect(dbapi_con, connection_record):
      print("新建 DBAPI 连接：", dbapi_con)
      cursor = dbapi_con.cursor()
  
      # 对适配后的 DBAPI 连接 / 游标使用同步风格 API
      cursor.execute("select 'execute from event'")
      print(cursor.fetchone()[0])
  
  
  # 在所有 Engine 实例上注册 before_execute 事件
  @event.listens_for(Engine, "before_execute")
  def my_before_execute(
      conn,
      clauseelement,
      multiparams,
      params,
      execution_options,
  ):
      print("before execute!")
  
  
  async def go():
      async with engine.connect() as conn:
          await conn.execute(text("select 1"))
      await engine.dispose()
  
  
  asyncio.run(go())

  输出结果：

  新建 DBAPI 连接： <AdaptedConnection <asyncpg.connection.Connection object at 0x7f33f9b16960>>
  execute from event
  before execute!

• AsyncSession 上的 ORM 事件

  在此示例中，我们通过 AsyncSession.sync_session 属性，作为 SessionEvents 的目标:

  import asyncio
  
  from sqlalchemy import event
  from sqlalchemy import text
  from sqlalchemy.ext.asyncio import AsyncSession
  from sqlalchemy.ext.asyncio import create_async_engine
  from sqlalchemy.orm import Session
  
  engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost:5432/test")
  
  session = AsyncSession(engine)
  
  
  # 在 Session 实例上注册 before_commit 事件
  @event.listens_for(session.sync_session, "before_commit")
  def my_before_commit(session):
      print("before commit!")
  
      # 在 Session 上使用同步风格 API
      connection = session.connection()
  
      # 在 Connection 上使用同步风格 API
      result = connection.execute(text("select 'execute from event'"))
      print(result.first())
  
  
  # 在所有 Session 实例上注册 after_commit 事件
  @event.listens_for(Session, "after_commit")
  def my_after_commit(session):
      print("after commit!")
  
  
  async def go():
      await session.execute(text("select 1"))
      await session.commit()
  
      await session.close()
      await engine.dispose()
  
  
  asyncio.run(go())

  输出结果：

  before commit!
  execute from event
  after commit!

• 在 async_sessionmaker 上的 ORM 事件

  在此用例中，我们以 sessionmaker 作为事件目标， 然后通过 async_sessionmaker.sync_session_class 参数， 将其赋值给 async_sessionmaker:

  import asyncio
  
  from sqlalchemy import event
  from sqlalchemy.ext.asyncio import async_sessionmaker
  from sqlalchemy.orm import sessionmaker
  
  sync_maker = sessionmaker()
  maker = async_sessionmaker(sync_session_class=sync_maker)
  
  
  @event.listens_for(sync_maker, "before_commit")
  def before_commit(session):
      print("before commit")
  
  
  async def main():
      async_session = maker()
  
      await async_session.commit()
  
  
  asyncio.run(main())

  输出结果：

  before commit

英文

Some examples of sync style event handlers associated with async-facing API constructs are illustrated below:

• Core Events on AsyncEngine

  In this example, we access the AsyncEngine.sync_engine attribute of AsyncEngine as the target for ConnectionEvents and PoolEvents:

  import asyncio
  
  from sqlalchemy import event
  from sqlalchemy import text
  from sqlalchemy.engine import Engine
  from sqlalchemy.ext.asyncio import create_async_engine
  
  engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost:5432/test")
  
  
  # connect event on instance of Engine
  @event.listens_for(engine.sync_engine, "connect")
  def my_on_connect(dbapi_con, connection_record):
      print("New DBAPI connection:", dbapi_con)
      cursor = dbapi_con.cursor()
  
      # sync style API use for adapted DBAPI connection / cursor
      cursor.execute("select 'execute from event'")
      print(cursor.fetchone()[0])
  
  
  # before_execute event on all Engine instances
  @event.listens_for(Engine, "before_execute")
  def my_before_execute(
      conn,
      clauseelement,
      multiparams,
      params,
      execution_options,
  ):
      print("before execute!")
  
  
  async def go():
      async with engine.connect() as conn:
          await conn.execute(text("select 1"))
      await engine.dispose()
  
  
  asyncio.run(go())

  Output:

  New DBAPI connection: <AdaptedConnection <asyncpg.connection.Connection object at 0x7f33f9b16960>>
  execute from event
  before execute!

• ORM Events on AsyncSession

  In this example, we access AsyncSession.sync_session as the target for SessionEvents:

  import asyncio
  
  from sqlalchemy import event
  from sqlalchemy import text
  from sqlalchemy.ext.asyncio import AsyncSession
  from sqlalchemy.ext.asyncio import create_async_engine
  from sqlalchemy.orm import Session
  
  engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost:5432/test")
  
  session = AsyncSession(engine)
  
  
  # before_commit event on instance of Session
  @event.listens_for(session.sync_session, "before_commit")
  def my_before_commit(session):
      print("before commit!")
  
      # sync style API use on Session
      connection = session.connection()
  
      # sync style API use on Connection
      result = connection.execute(text("select 'execute from event'"))
      print(result.first())
  
  
  # after_commit event on all Session instances
  @event.listens_for(Session, "after_commit")
  def my_after_commit(session):
      print("after commit!")
  
  
  async def go():
      await session.execute(text("select 1"))
      await session.commit()
  
      await session.close()
      await engine.dispose()
  
  
  asyncio.run(go())

  Output:

  before commit!
  execute from event
  after commit!

• ORM Events on async_sessionmaker

  For this use case, we make a sessionmaker as the event target, then assign it to the async_sessionmaker using the async_sessionmaker.sync_session_class parameter:

  import asyncio
  
  from sqlalchemy import event
  from sqlalchemy.ext.asyncio import async_sessionmaker
  from sqlalchemy.orm import sessionmaker
  
  sync_maker = sessionmaker()
  maker = async_sessionmaker(sync_session_class=sync_maker)
  
  
  @event.listens_for(sync_maker, "before_commit")
  def before_commit(session):
      print("before commit")
  
  
  async def main():
      async_session = maker()
  
      await async_session.commit()
  
  
  asyncio.run(main())

  Output:

  before commit

在连接池和其他事件中使用仅可等待的驱动程序方法

Using awaitable-only driver methods in connection pool and other events

中文

如上节所述，诸如围绕 PoolEvents 的事件处理器会接收到一个同步风格的 “DBAPI” 连接， 这是 SQLAlchemy asyncio 方言提供的一个包装对象，用于将底层 asyncio “驱动”连接适配为 SQLAlchemy 内部可以使用的形式。 当用户自定义的事件处理器实现需要直接使用底层驱动连接、 并调用其仅支持 await 的方法时，就会出现一种特殊用例。一个典型的例子是 asyncpg 驱动所提供的 .set_type_codec() 方法。

为支持这一用例，SQLAlchemy 的 AdaptedConnection 类提供了一个方法 AdaptedConnection.run_async()，它允许在事件处理器或其他 SQLAlchemy 内部逻辑的 “同步”上下文中调用一个可等待（awaitable）的函数。该方法的设计与 AsyncConnection.run_sync() 相对应，后者允许在异步上下文中运行同步方法。

AdaptedConnection.run_async() 应当接收一个函数，该函数接受最底层的 “驱动”连接作为唯一参数， 并返回一个可等待对象。该函数本身不需要声明为 async；完全可以是一个 Python 的 lambda 表达式， 因为返回的 awaitable 会在之后被调用:

from sqlalchemy import event
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(...)


@event.listens_for(engine.sync_engine, "connect")
def register_custom_types(dbapi_connection, *args):
    dbapi_connection.run_async(
        lambda connection: connection.set_type_codec(
            "MyCustomType",
            encoder,
            decoder,  # ...
        )
    )

在上述示例中，传递给 register_custom_types 事件处理器的对象是一个 AdaptedConnection 实例， 它为底层仅支持异步的驱动级连接对象提供了类 DBAPI 的接口。 AdaptedConnection.run_async() 方法则提供了一个 awaitable 环境，可对底层驱动连接进行操作。

在 1.4.30 版本加入.

英文

As discussed in the above section, event handlers such as those oriented around the PoolEvents event handlers receive a sync-style “DBAPI” connection, which is a wrapper object supplied by SQLAlchemy asyncio dialects to adapt the underlying asyncio “driver” connection into one that can be used by SQLAlchemy’s internals. A special use case arises when the user-defined implementation for such an event handler needs to make use of the ultimate “driver” connection directly, using awaitable only methods on that driver connection. One such example is the .set_type_codec() method supplied by the asyncpg driver.

To accommodate this use case, SQLAlchemy’s AdaptedConnection class provides a method AdaptedConnection.run_async() that allows an awaitable function to be invoked within the “synchronous” context of an event handler or other SQLAlchemy internal. This method is directly analogous to the AsyncConnection.run_sync() method that allows a sync-style method to run under async.

AdaptedConnection.run_async() should be passed a function that will accept the innermost “driver” connection as a single argument, and return an awaitable that will be invoked by the AdaptedConnection.run_async() method. The given function itself does not need to be declared as async; it’s perfectly fine for it to be a Python lambda:, as the return awaitable value will be invoked after being returned:

from sqlalchemy import event
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(...)


@event.listens_for(engine.sync_engine, "connect")
def register_custom_types(dbapi_connection, *args):
    dbapi_connection.run_async(
        lambda connection: connection.set_type_codec(
            "MyCustomType",
            encoder,
            decoder,  # ...
        )
    )

Above, the object passed to the register_custom_types event handler is an instance of AdaptedConnection, which provides a DBAPI-like interface to an underlying async-only driver-level connection object. The AdaptedConnection.run_async() method then provides access to an awaitable environment where the underlying driver level connection may be acted upon.

在 1.4.30 版本加入.

使用多个 asyncio 事件循环

Using multiple asyncio event loops

中文

如果一个应用程序使用了多个事件循环（例如在异步编程中不常见地与多线程结合使用）， 在使用默认连接池实现的情况下，不应在不同事件循环之间共享同一个 AsyncEngine。

如果需要将 AsyncEngine 从一个事件循环传递到另一个， 则应在重新使用前调用 AsyncEngine.dispose() 方法。 若未正确处理，可能会导致 RuntimeError 异常， 类似于 Task <Task pending ...> got Future attached to a different loop。

如果必须在不同事件循环之间共享同一个引擎，应通过配置 NullPool 禁用连接池，以避免任何连接被重复使用:

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@host/dbname",
    poolclass=NullPool,
)

英文

An application that makes use of multiple event loops, for example in the uncommon case of combining asyncio with multithreading, should not share the same AsyncEngine with different event loops when using the default pool implementation.

If an AsyncEngine is be passed from one event loop to another, the method AsyncEngine.dispose() should be called before it’s re-used on a new event loop. Failing to do so may lead to a RuntimeError along the lines of Task <Task pending ...> got Future attached to a different loop

If the same engine must be shared between different loop, it should be configured to disable pooling using NullPool, preventing the Engine from using any connection more than once:

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@host/dbname",
    poolclass=NullPool,
)

使用 asyncio 作用域会话

Using asyncio scoped session

中文

在线程模式的 SQLAlchemy 中所使用的 “scoped session” 模式（通过 scoped_session 对象实现）， 在 asyncio 中也有适配版本，即 async_scoped_session。

小技巧

SQLAlchemy 通常 不推荐 在新开发中使用 “scoped” 模式，因为它依赖于可变的全局状态， 且在每个线程或任务工作完成后还必须显式清理这些状态。 特别是在使用 asyncio 时，直接将 AsyncSession 作为参数传递给需要它的 awaitable 函数通常是更好的做法。

在使用 async_scoped_session 时，由于 asyncio 上下文中没有 “线程本地” 的概念， 必须为其构造函数提供 “scopefunc” 参数。以下示例演示了使用 asyncio.current_task() 实现此目的:

from asyncio import current_task

from sqlalchemy.ext.asyncio import (
    async_scoped_session,
    async_sessionmaker,
)

async_session_factory = async_sessionmaker(
    some_async_engine,
    expire_on_commit=False,
)
AsyncScopedSession = async_scoped_session(
    async_session_factory,
    scopefunc=current_task,
)
some_async_session = AsyncScopedSession()

警告

async_scoped_session 所使用的 “scopefunc” 会在一个任务中 被多次调用，每次访问底层 AsyncSession 时都会执行一次。 因此，该函数应具有 幂等性 且尽可能轻量，不应尝试创建或修改任何状态（例如建立回调等）。

警告

使用 current_task() 作为作用域键时，必须在最外层 awaitable 作用域内调用 async_scoped_session.remove() 方法，以确保任务完成时将作用域键从注册表中移除， 否则任务句柄和 AsyncSession 实例会保留在内存中，导致内存泄漏。 下面的示例展示了正确使用 async_scoped_session.remove() 的方式。

async_scoped_session 实现了与 scoped_session 类似的 代理行为， 因此它可以直接被当作 AsyncSession 使用，只需注意在调用时添加 await， 包括调用 async_scoped_session.remove() 方法:

async def some_function(some_async_session, some_object):
    # 直接使用 AsyncSession
    some_async_session.add(some_object)

    # 通过上下文代理使用 AsyncSession
    await AsyncScopedSession.commit()

    # 移除当前上下文代理的 AsyncSession
    await AsyncScopedSession.remove()

在 1.4.19 版本加入.

英文

The “scoped session” pattern used in threaded SQLAlchemy with the scoped_session object is also available in asyncio, using an adapted version called async_scoped_session.

小技巧

SQLAlchemy generally does not recommend the “scoped” pattern for new development as it relies upon mutable global state that must also be explicitly torn down when work within the thread or task is complete. Particularly when using asyncio, it’s likely a better idea to pass the AsyncSession directly to the awaitable functions that need it.

When using async_scoped_session, as there’s no “thread-local” concept in the asyncio context, the “scopefunc” parameter must be provided to the constructor. The example below illustrates using the asyncio.current_task() function for this purpose:

from asyncio import current_task

from sqlalchemy.ext.asyncio import (
    async_scoped_session,
    async_sessionmaker,
)

async_session_factory = async_sessionmaker(
    some_async_engine,
    expire_on_commit=False,
)
AsyncScopedSession = async_scoped_session(
    async_session_factory,
    scopefunc=current_task,
)
some_async_session = AsyncScopedSession()

警告

The “scopefunc” used by async_scoped_session is invoked an arbitrary number of times within a task, once for each time the underlying AsyncSession is accessed. The function should therefore be idempotent and lightweight, and should not attempt to create or mutate any state, such as establishing callbacks, etc.

警告

Using current_task() for the “key” in the scope requires that the async_scoped_session.remove() method is called from within the outermost awaitable, to ensure the key is removed from the registry when the task is complete, otherwise the task handle as well as the AsyncSession will remain in memory, essentially creating a memory leak. See the following example which illustrates the correct use of async_scoped_session.remove().

async_scoped_session includes proxy behavior similar to that of scoped_session, which means it can be treated as a AsyncSession directly, keeping in mind that the usual await keywords are necessary, including for the async_scoped_session.remove() method:

async def some_function(some_async_session, some_object):
    # use the AsyncSession directly
    some_async_session.add(some_object)

    # use the AsyncSession via the context-local proxy
    await AsyncScopedSession.commit()

    # "remove" the current proxied AsyncSession for the local context
    await AsyncScopedSession.remove()

在 1.4.19 版本加入.

使用检查器检查架构对象

Using the Inspector to inspect schema objects

中文

SQLAlchemy 还没有提供 Inspector （在 使用检查器的细粒度反射 中介绍）的 asyncio 版本， 但是可以通过利用 AsyncConnection 的 AsyncConnection.run_sync() 方法在 asyncio 上下文中使用现有接口:

import asyncio

from sqlalchemy import inspect
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost/test")


def use_inspector(conn):
    inspector = inspect(conn)
    # use the inspector
    print(inspector.get_view_names())
    # return any value to the caller
    return inspector.get_table_names()


async def async_main():
    async with engine.connect() as conn:
        tables = await conn.run_sync(use_inspector)


asyncio.run(async_main())

英文

SQLAlchemy does not yet offer an asyncio version of the Inspector (introduced at 使用检查器的细粒度反射), however the existing interface may be used in an asyncio context by leveraging the AsyncConnection.run_sync() method of AsyncConnection:

import asyncio

from sqlalchemy import inspect
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost/test")


def use_inspector(conn):
    inspector = inspect(conn)
    # use the inspector
    print(inspector.get_view_names())
    # return any value to the caller
    return inspector.get_table_names()


async def async_main():
    async with engine.connect() as conn:
        tables = await conn.run_sync(use_inspector)


asyncio.run(async_main())

参见

反射数据库对象

运行时审查 API

引擎 API 文档

Engine API Documentation

Object Name	Description
async_engine_from_config(configuration[, prefix], **kwargs)	Create a new AsyncEngine instance using a configuration dictionary.
AsyncConnection	An asyncio proxy for a Connection.
AsyncEngine	An asyncio proxy for a Engine.
AsyncTransaction	An asyncio proxy for a Transaction.
create_async_engine(url, **kw)	Create a new async engine instance.
create_async_pool_from_url(url, **kwargs)	Create a new async engine instance.

function sqlalchemy.ext.asyncio.create_async_engine(url: str | URL, **kw: Any) → AsyncEngine

Create a new async engine instance.

Arguments passed to create_async_engine() are mostly identical to those passed to the create_engine() function. The specified dialect must be an asyncio-compatible dialect such as asyncpg.

在 1.4 版本加入.

参数:

async_creator –

an async callable which returns a driver-level asyncio connection. If given, the function should take no arguments, and return a new asyncio connection from the underlying asyncio database driver; the connection will be wrapped in the appropriate structures to be used with the AsyncEngine. Note that the parameters specified in the URL are not applied here, and the creator function should use its own connection parameters.

This parameter is the asyncio equivalent of the create_engine.creator parameter of the create_engine() function.

在 2.0.16 版本加入.

function sqlalchemy.ext.asyncio.async_engine_from_config(configuration: Dict[str, Any], prefix: str = 'sqlalchemy.', **kwargs: Any) → AsyncEngine

Create a new AsyncEngine instance using a configuration dictionary.

This function is analogous to the engine_from_config() function in SQLAlchemy Core, except that the requested dialect must be an asyncio-compatible dialect such as asyncpg. The argument signature of the function is identical to that of engine_from_config().

在 1.4.29 版本加入.

function sqlalchemy.ext.asyncio.create_async_pool_from_url(url: str | URL, **kwargs: Any) → Pool

Create a new async engine instance.

Arguments passed to create_async_pool_from_url() are mostly identical to those passed to the create_pool_from_url() function. The specified dialect must be an asyncio-compatible dialect such as asyncpg.

在 2.0.10 版本加入.

class sqlalchemy.ext.asyncio.AsyncEngine

An asyncio proxy for a Engine.

AsyncEngine is acquired using the create_async_engine() function:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:pass@host/dbname")

在 1.4 版本加入.

Members

begin(), clear_compiled_cache(), connect(), dialect, dispose(), driver, echo, engine, execution_options(), get_execution_options(), name, pool, raw_connection(), sync_engine, update_execution_options(), url

Class signature

class sqlalchemy.ext.asyncio.AsyncEngine (sqlalchemy.ext.asyncio.base.ProxyComparable, sqlalchemy.ext.asyncio.AsyncConnectable)

method sqlalchemy.ext.asyncio.AsyncEngine.begin() → AsyncIterator[AsyncConnection]

Return a context manager which when entered will deliver an AsyncConnection with an AsyncTransaction established.

E.g.:

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

method sqlalchemy.ext.asyncio.AsyncEngine.clear_compiled_cache() → None

Clear the compiled cache associated with the dialect.

Proxied for the Engine class on behalf of the AsyncEngine class.

This applies only to the built-in cache that is established via the create_engine.query_cache_size parameter. It will not impact any dictionary caches that were passed via the Connection.execution_options.compiled_cache parameter.

在 1.4 版本加入.

method sqlalchemy.ext.asyncio.AsyncEngine.connect() → AsyncConnection

Return an AsyncConnection object.

The AsyncConnection will procure a database connection from the underlying connection pool when it is entered as an async context manager:

async with async_engine.connect() as conn:
    result = await conn.execute(select(user_table))

The AsyncConnection may also be started outside of a context manager by invoking its AsyncConnection.start() method.

attribute sqlalchemy.ext.asyncio.AsyncEngine.dialect

Proxy for the Engine.dialect attribute on behalf of the AsyncEngine class.

method sqlalchemy.ext.asyncio.AsyncEngine.async dispose(close: bool = True) → None

Dispose of the connection pool used by this AsyncEngine.

参数:

close –

if left at its default of True, has the effect of fully closing all currently checked in database connections. Connections that are still checked out will not be closed, however they will no longer be associated with this Engine, so when they are closed individually, eventually the Pool which they are associated with will be garbage collected and they will be closed out fully, if not already closed on checkin.

If set to False, the previous connection pool is de-referenced, and otherwise not touched in any way.

参见

Engine.dispose()

attribute sqlalchemy.ext.asyncio.AsyncEngine.driver

Driver name of the Dialect in use by this Engine.

Proxied for the Engine class on behalf of the AsyncEngine class.

attribute sqlalchemy.ext.asyncio.AsyncEngine.echo

When True, enable log output for this element.

Proxied for the Engine class on behalf of the AsyncEngine class.

This has the effect of setting the Python logging level for the namespace of this element’s class and object reference. A value of boolean True indicates that the loglevel logging.INFO will be set for the logger, whereas the string value debug will set the loglevel to logging.DEBUG.

attribute sqlalchemy.ext.asyncio.AsyncEngine.engine

Returns this Engine.

Proxied for the Engine class on behalf of the AsyncEngine class.

Used for legacy schemes that accept Connection / Engine objects within the same variable.

method sqlalchemy.ext.asyncio.AsyncEngine.execution_options(**opt: Any) → AsyncEngine

Return a new AsyncEngine that will provide AsyncConnection objects with the given execution options.

Proxied from Engine.execution_options(). See that method for details.

method sqlalchemy.ext.asyncio.AsyncEngine.get_execution_options() → _ExecuteOptions

Get the non-SQL options which will take effect during execution.

Proxied for the Engine class on behalf of the AsyncEngine class.

参见

Engine.execution_options()

attribute sqlalchemy.ext.asyncio.AsyncEngine.name

String name of the Dialect in use by this Engine.

Proxied for the Engine class on behalf of the AsyncEngine class.

attribute sqlalchemy.ext.asyncio.AsyncEngine.pool

Proxy for the Engine.pool attribute on behalf of the AsyncEngine class.

method sqlalchemy.ext.asyncio.AsyncEngine.async raw_connection() → PoolProxiedConnection

Return a “raw” DBAPI connection from the connection pool.

参见

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

attribute sqlalchemy.ext.asyncio.AsyncEngine.sync_engine: Engine

Reference to the sync-style Engine this AsyncEngine proxies requests towards.

This instance can be used as an event target.

参见

使用 asyncio 扩展的事件

method sqlalchemy.ext.asyncio.AsyncEngine.update_execution_options(**opt: Any) → None

Update the default execution_options dictionary of this Engine.

Proxied for the Engine class on behalf of the AsyncEngine class.

The given keys/values in **opt are added to the default execution options that will be used for all connections. The initial contents of this dictionary can be sent via the execution_options parameter to create_engine().

参见

Connection.execution_options()

Engine.execution_options()

attribute sqlalchemy.ext.asyncio.AsyncEngine.url

Proxy for the Engine.url attribute on behalf of the AsyncEngine class.

class sqlalchemy.ext.asyncio.AsyncConnection

An asyncio proxy for a Connection.

AsyncConnection is acquired using the AsyncEngine.connect() method of AsyncEngine:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:pass@host/dbname")

async with engine.connect() as conn:
    result = await conn.execute(select(table))

在 1.4 版本加入.

Members

aclose(), begin(), begin_nested(), close(), closed, commit(), connection, default_isolation_level, dialect, exec_driver_sql(), execute(), execution_options(), get_nested_transaction(), get_raw_connection(), get_transaction(), in_nested_transaction(), in_transaction(), info, invalidate(), invalidated, rollback(), run_sync(), scalar(), scalars(), start(), stream(), stream_scalars(), sync_connection, sync_engine

Class signature

class sqlalchemy.ext.asyncio.AsyncConnection (sqlalchemy.ext.asyncio.base.ProxyComparable, sqlalchemy.ext.asyncio.base.StartableContext, sqlalchemy.ext.asyncio.AsyncConnectable)

method sqlalchemy.ext.asyncio.AsyncConnection.async aclose() → None

A synonym for AsyncConnection.close().

The AsyncConnection.aclose() name is specifically to support the Python standard library @contextlib.aclosing context manager function.

在 2.0.20 版本加入.

method sqlalchemy.ext.asyncio.AsyncConnection.begin() → AsyncTransaction

Begin a transaction prior to autobegin occurring.

method sqlalchemy.ext.asyncio.AsyncConnection.begin_nested() → AsyncTransaction

Begin a nested transaction and return a transaction handle.

method sqlalchemy.ext.asyncio.AsyncConnection.async close() → None

Close this AsyncConnection.

This has the effect of also rolling back the transaction if one is in place.

attribute sqlalchemy.ext.asyncio.AsyncConnection.closed

Return True if this connection is closed.

Proxied for the Connection class on behalf of the AsyncConnection class.

method sqlalchemy.ext.asyncio.AsyncConnection.async commit() → None

Commit the transaction that is currently in progress.

This method commits the current transaction if one has been started. If no transaction was started, the method has no effect, assuming the connection is in a non-invalidated state.

A transaction is begun on a Connection automatically whenever a statement is first executed, or when the Connection.begin() method is called.

attribute sqlalchemy.ext.asyncio.AsyncConnection.connection

Not implemented for async; call AsyncConnection.get_raw_connection().

attribute sqlalchemy.ext.asyncio.AsyncConnection.default_isolation_level

The initial-connection time isolation level associated with the Dialect in use.

Proxied for the Connection class on behalf of the AsyncConnection class.

This value is independent of the Connection.execution_options.isolation_level and Engine.execution_options.isolation_level execution options, and is determined by the Dialect when the first connection is created, by performing a SQL query against the database for the current isolation level before any additional commands have been emitted.

Calling this accessor does not invoke any new SQL queries.

参见

Connection.get_isolation_level() - view current actual isolation level

create_engine.isolation_level - set per Engine isolation level

Connection.execution_options.isolation_level - set per Connection isolation level

attribute sqlalchemy.ext.asyncio.AsyncConnection.dialect

Proxy for the Connection.dialect attribute on behalf of the AsyncConnection class.

method sqlalchemy.ext.asyncio.AsyncConnection.async exec_driver_sql(statement: str, parameters: _DBAPIAnyExecuteParams | None = None, execution_options: CoreExecuteOptionsParameter | None = None) → CursorResult[Any]

Executes a driver-level SQL string and return buffered Result.

method sqlalchemy.ext.asyncio.AsyncConnection.async execute(statement: Executable, parameters: _CoreAnyExecuteParams | None = None, *, execution_options: CoreExecuteOptionsParameter | None = None) → CursorResult[Unpack[TupleAny]]

Executes a SQL statement construct and return a buffered Result.

参数:

• object –

  The statement to be executed. This is always an object that is in both the ClauseElement and Executable hierarchies, including:

  • Select

  • Insert, Update, Delete

  • TextClause and TextualSelect

  • DDL and objects which inherit from ExecutableDDLElement

• parameters – parameters which will be bound into the statement. This may be either a dictionary of parameter names to values, or a mutable sequence (e.g. a list) of dictionaries. When a list of dictionaries is passed, the underlying statement execution will make use of the DBAPI cursor.executemany() method. When a single dictionary is passed, the DBAPI cursor.execute() method will be used.

• execution_options – optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by Connection.execution_options().

返回:

a Result object.

method sqlalchemy.ext.asyncio.AsyncConnection.async execution_options(**opt: Any) → AsyncConnection

Set non-SQL options for the connection which take effect during execution.

This returns this AsyncConnection object with the new options added.

See Connection.execution_options() for full details on this method.

method sqlalchemy.ext.asyncio.AsyncConnection.get_nested_transaction() → AsyncTransaction | None

Return an AsyncTransaction representing the current nested (savepoint) transaction, if any.

This makes use of the underlying synchronous connection’s Connection.get_nested_transaction() method to get the current Transaction, which is then proxied in a new AsyncTransaction object.

在 1.4.0b2 版本加入.

method sqlalchemy.ext.asyncio.AsyncConnection.async get_raw_connection() → PoolProxiedConnection

Return the pooled DBAPI-level connection in use by this AsyncConnection.

This is a SQLAlchemy connection-pool proxied connection which then has the attribute _ConnectionFairy.driver_connection that refers to the actual driver connection. Its _ConnectionFairy.dbapi_connection refers instead to an AdaptedConnection instance that adapts the driver connection to the DBAPI protocol.

method sqlalchemy.ext.asyncio.AsyncConnection.get_transaction() → AsyncTransaction | None

Return an AsyncTransaction representing the current transaction, if any.

This makes use of the underlying synchronous connection’s Connection.get_transaction() method to get the current Transaction, which is then proxied in a new AsyncTransaction object.

在 1.4.0b2 版本加入.

method sqlalchemy.ext.asyncio.AsyncConnection.in_nested_transaction() → bool

Return True if a transaction is in progress.

在 1.4.0b2 版本加入.

method sqlalchemy.ext.asyncio.AsyncConnection.in_transaction() → bool

Return True if a transaction is in progress.

attribute sqlalchemy.ext.asyncio.AsyncConnection.info

Return the Connection.info dictionary of the underlying Connection.

This dictionary is freely writable for user-defined state to be associated with the database connection.

This attribute is only available if the AsyncConnection is currently connected. If the AsyncConnection.closed attribute is True, then accessing this attribute will raise ResourceClosedError.

在 1.4.0b2 版本加入.

method sqlalchemy.ext.asyncio.AsyncConnection.async invalidate(exception: BaseException | None = None) → None

Invalidate the underlying DBAPI connection associated with this Connection.

See the method Connection.invalidate() for full detail on this method.

attribute sqlalchemy.ext.asyncio.AsyncConnection.invalidated

Return True if this connection was invalidated.

Proxied for the Connection class on behalf of the AsyncConnection class.

This does not indicate whether or not the connection was invalidated at the pool level, however

method sqlalchemy.ext.asyncio.AsyncConnection.async rollback() → None

Roll back the transaction that is currently in progress.

This method rolls back the current transaction if one has been started. If no transaction was started, the method has no effect. If a transaction was started and the connection is in an invalidated state, the transaction is cleared using this method.

A transaction is begun on a Connection automatically whenever a statement is first executed, or when the Connection.begin() method is called.

method sqlalchemy.ext.asyncio.AsyncConnection.async run_sync(fn: ~typing.Callable[[~typing.Concatenate[~sqlalchemy.engine.base.Connection, ~_P]], ~sqlalchemy.ext.asyncio.engine._T], *arg: ~typing.~_P, **kw: ~typing.~_P) → _T

Invoke the given synchronous (i.e. not async) callable, passing a synchronous-style Connection as the first argument.

This method allows traditional synchronous SQLAlchemy functions to run within the context of an asyncio application.

E.g.:

def do_something_with_core(conn: Connection, arg1: int, arg2: str) -> str:
    """A synchronous function that does not require awaiting

    :param conn: a Core SQLAlchemy Connection, used synchronously

    :return: an optional return value is supported

    """
    conn.execute(some_table.insert().values(int_col=arg1, str_col=arg2))
    return "success"


async def do_something_async(async_engine: AsyncEngine) -> None:
    """an async function that uses awaiting"""

    async with async_engine.begin() as async_conn:
        # run do_something_with_core() with a sync-style
        # Connection, proxied into an awaitable
        return_code = await async_conn.run_sync(
            do_something_with_core, 5, "strval"
        )
        print(return_code)

This method maintains the asyncio event loop all the way through to the database connection by running the given callable in a specially instrumented greenlet.

The most rudimentary use of AsyncConnection.run_sync() is to invoke methods such as MetaData.create_all(), given an AsyncConnection that needs to be provided to MetaData.create_all() as a Connection object:

# run metadata.create_all(conn) with a sync-style Connection,
# proxied into an awaitable
with async_engine.begin() as conn:
    await conn.run_sync(metadata.create_all)

备注

The provided callable is invoked inline within the asyncio event loop, and will block on traditional IO calls. IO within this callable should only call into SQLAlchemy’s asyncio database APIs which will be properly adapted to the greenlet context.

参见

AsyncSession.run_sync()

在 asyncio 下运行同步方法和函数

method sqlalchemy.ext.asyncio.AsyncConnection.async scalar(statement: Executable, parameters: _CoreSingleExecuteParams | None = None, *, execution_options: CoreExecuteOptionsParameter | None = None) → Any

Executes a SQL statement construct and returns a scalar object.

This method is shorthand for invoking the Result.scalar() method after invoking the Connection.execute() method. Parameters are equivalent.

返回:

a scalar Python value representing the first column of the first row returned.

method sqlalchemy.ext.asyncio.AsyncConnection.async scalars(statement: Executable, parameters: _CoreAnyExecuteParams | None = None, *, execution_options: CoreExecuteOptionsParameter | None = None) → ScalarResult[Any]

Executes a SQL statement construct and returns a scalar objects.

This method is shorthand for invoking the Result.scalars() method after invoking the Connection.execute() method. Parameters are equivalent.

返回:

a ScalarResult object.

在 1.4.24 版本加入.

method sqlalchemy.ext.asyncio.AsyncConnection.async start(is_ctxmanager: bool = False) → AsyncConnection

Start this AsyncConnection object’s context outside of using a Python with: block.

method sqlalchemy.ext.asyncio.AsyncConnection.stream(statement: Executable, parameters: _CoreAnyExecuteParams | None = None, *, execution_options: CoreExecuteOptionsParameter | None = None) → AsyncIterator[AsyncResult[Unpack[TupleAny]]]

Execute a statement and return an awaitable yielding a AsyncResult object.

E.g.:

result = await conn.stream(stmt)
async for row in result:
    print(f"{row}")

The AsyncConnection.stream() method supports optional context manager use against the AsyncResult object, as in:

async with conn.stream(stmt) as result:
    async for row in result:
        print(f"{row}")

In the above pattern, the AsyncResult.close() method is invoked unconditionally, even if the iterator is interrupted by an exception throw. Context manager use remains optional, however, and the function may be called in either an async with fn(): or await fn() style.

在 2.0.0b3 版本加入: added context manager support

返回:

an awaitable object that will yield an AsyncResult object.

参见

AsyncConnection.stream_scalars()

method sqlalchemy.ext.asyncio.AsyncConnection.stream_scalars(statement: Executable, parameters: _CoreSingleExecuteParams | None = None, *, execution_options: CoreExecuteOptionsParameter | None = None) → AsyncIterator[AsyncScalarResult[Any]]

Execute a statement and return an awaitable yielding a AsyncScalarResult object.

E.g.:

result = await conn.stream_scalars(stmt)
async for scalar in result:
    print(f"{scalar}")

This method is shorthand for invoking the AsyncResult.scalars() method after invoking the Connection.stream() method. Parameters are equivalent.

The AsyncConnection.stream_scalars() method supports optional context manager use against the AsyncScalarResult object, as in:

async with conn.stream_scalars(stmt) as result:
    async for scalar in result:
        print(f"{scalar}")

In the above pattern, the AsyncScalarResult.close() method is invoked unconditionally, even if the iterator is interrupted by an exception throw. Context manager use remains optional, however, and the function may be called in either an async with fn(): or await fn() style.

在 2.0.0b3 版本加入: added context manager support

返回:

an awaitable object that will yield an AsyncScalarResult object.

在 1.4.24 版本加入.

参见

AsyncConnection.stream()

attribute sqlalchemy.ext.asyncio.AsyncConnection.sync_connection: Connection | None

Reference to the sync-style Connection this AsyncConnection proxies requests towards.

This instance can be used as an event target.

参见

使用 asyncio 扩展的事件

attribute sqlalchemy.ext.asyncio.AsyncConnection.sync_engine: Engine

Reference to the sync-style Engine this AsyncConnection is associated with via its underlying Connection.

This instance can be used as an event target.

参见

使用 asyncio 扩展的事件

class sqlalchemy.ext.asyncio.AsyncTransaction

An asyncio proxy for a Transaction.

Members

close(), commit(), rollback(), start()

Class signature

class sqlalchemy.ext.asyncio.AsyncTransaction (sqlalchemy.ext.asyncio.base.ProxyComparable, sqlalchemy.ext.asyncio.base.StartableContext)

method sqlalchemy.ext.asyncio.AsyncTransaction.async close() → None

Close this AsyncTransaction.

If this transaction is the base transaction in a begin/commit nesting, the transaction will rollback(). Otherwise, the method returns.

This is used to cancel a Transaction without affecting the scope of an enclosing transaction.

method sqlalchemy.ext.asyncio.AsyncTransaction.async commit() → None

Commit this AsyncTransaction.

method sqlalchemy.ext.asyncio.AsyncTransaction.async rollback() → None

Roll back this AsyncTransaction.

method sqlalchemy.ext.asyncio.AsyncTransaction.async start(is_ctxmanager: bool = False) → AsyncTransaction

Start this AsyncTransaction object’s context outside of using a Python with: block.

结果集 API 文档

Result Set API Documentation

The AsyncResult object is an async-adapted version of the Result object. It is only returned when using the AsyncConnection.stream() or AsyncSession.stream() methods, which return a result object that is on top of an active database cursor.

Object Name	Description
AsyncMappingResult	A wrapper for a AsyncResult that returns dictionary values rather than Row values.
AsyncResult	An asyncio wrapper around a Result object.
AsyncScalarResult	A wrapper for a AsyncResult that returns scalar values rather than Row values.
AsyncTupleResult	A AsyncResult that’s typed as returning plain Python tuples instead of rows.

class sqlalchemy.ext.asyncio.AsyncResult

An asyncio wrapper around a Result object.

The AsyncResult only applies to statement executions that use a server-side cursor. It is returned only from the AsyncConnection.stream() and AsyncSession.stream() methods.

备注

As is the case with Result, this object is used for ORM results returned by AsyncSession.execute(), which can yield instances of ORM mapped objects either individually or within tuple-like rows. Note that these result objects do not deduplicate instances or rows automatically as is the case with the legacy Query object. For in-Python de-duplication of instances or rows, use the AsyncResult.unique() modifier method.

在 1.4 版本加入.

Members

all(), close(), closed, columns(), fetchall(), fetchmany(), fetchone(), first(), freeze(), keys(), mappings(), one(), one_or_none(), partitions(), scalar(), scalar_one(), scalar_one_or_none(), scalars(), t, tuples(), unique(), yield_per()

Class signature

class sqlalchemy.ext.asyncio.AsyncResult (sqlalchemy.engine._WithKeys, sqlalchemy.ext.asyncio.AsyncCommon)

method sqlalchemy.ext.asyncio.AsyncResult.async all() → Sequence[Row[Unpack[_Ts]]]

Return all rows in a list.

Closes the result set after invocation. Subsequent invocations will return an empty list.

返回:

a list of Row objects.

method sqlalchemy.ext.asyncio.AsyncResult.async close() → None

inherited from the AsyncCommon.close() method of AsyncCommon

Close this result.

attribute sqlalchemy.ext.asyncio.AsyncResult.closed

inherited from the AsyncCommon.closed attribute of AsyncCommon

proxies the .closed attribute of the underlying result object, if any, else raises AttributeError.

在 2.0.0b3 版本加入.

method sqlalchemy.ext.asyncio.AsyncResult.columns(*col_expressions: _KeyIndexType) → Self

Establish the columns that should be returned in each row.

Refer to Result.columns() in the synchronous SQLAlchemy API for a complete behavioral description.

method sqlalchemy.ext.asyncio.AsyncResult.async fetchall() → Sequence[Row[Unpack[_Ts]]]

A synonym for the AsyncResult.all() method.

在 2.0 版本加入.

method sqlalchemy.ext.asyncio.AsyncResult.async fetchmany(size: int | None = None) → Sequence[Row[Unpack[_Ts]]]

Fetch many rows.

When all rows are exhausted, returns an empty list.

This method is provided for backwards compatibility with SQLAlchemy 1.x.x.

To fetch rows in groups, use the AsyncResult.partitions() method.

返回:

a list of Row objects.

参见

AsyncResult.partitions()

method sqlalchemy.ext.asyncio.AsyncResult.async fetchone() → Row[Unpack[_Ts]] | None

Fetch one row.

When all rows are exhausted, returns None.

This method is provided for backwards compatibility with SQLAlchemy 1.x.x.

To fetch the first row of a result only, use the AsyncResult.first() method. To iterate through all rows, iterate the AsyncResult object directly.

返回:

a Row object if no filters are applied, or None if no rows remain.

method sqlalchemy.ext.asyncio.AsyncResult.async first() → Row[Unpack[_Ts]] | None

Fetch the first row or None if no row is present.

Closes the result set and discards remaining rows.

备注

This method returns one row, e.g. tuple, by default. To return exactly one single scalar value, that is, the first column of the first row, use the AsyncResult.scalar() method, or combine AsyncResult.scalars() and AsyncResult.first().

Additionally, in contrast to the behavior of the legacy ORM Query.first() method, no limit is applied to the SQL query which was invoked to produce this AsyncResult; for a DBAPI driver that buffers results in memory before yielding rows, all rows will be sent to the Python process and all but the first row will be discarded.

参见

ORM Query Unified with Core Select

返回:

a Row object, or None if no rows remain.

参见

AsyncResult.scalar()

AsyncResult.one()

method sqlalchemy.ext.asyncio.AsyncResult.async freeze() → FrozenResult[Unpack[_Ts]]

Return a callable object that will produce copies of this AsyncResult when invoked.

The callable object returned is an instance of FrozenResult.

This is used for result set caching. The method must be called on the result when it has been unconsumed, and calling the method will consume the result fully. When the FrozenResult is retrieved from a cache, it can be called any number of times where it will produce a new Result object each time against its stored set of rows.

参见

重新执行语句 - example usage within the ORM to implement a result-set cache.

method sqlalchemy.ext.asyncio.AsyncResult.keys() → RMKeyView

inherited from the sqlalchemy.engine._WithKeys.keys method of sqlalchemy.engine._WithKeys

Return an iterable view which yields the string keys that would be represented by each Row.

The keys can represent the labels of the columns returned by a core statement or the names of the orm classes returned by an orm execution.

The view also can be tested for key containment using the Python in operator, which will test both for the string keys represented in the view, as well as for alternate keys such as column objects.

在 1.4 版本发生变更: a key view object is returned rather than a plain list.

method sqlalchemy.ext.asyncio.AsyncResult.mappings() → AsyncMappingResult

Apply a mappings filter to returned rows, returning an instance of AsyncMappingResult.

When this filter is applied, fetching rows will return RowMapping objects instead of Row objects.

返回:

a new AsyncMappingResult filtering object referring to the underlying Result object.

method sqlalchemy.ext.asyncio.AsyncResult.async one() → Row[Unpack[_Ts]]

Return exactly one row or raise an exception.

Raises NoResultFound if the result returns no rows, or MultipleResultsFound if multiple rows would be returned.

备注

This method returns one row, e.g. tuple, by default. To return exactly one single scalar value, that is, the first column of the first row, use the AsyncResult.scalar_one() method, or combine AsyncResult.scalars() and AsyncResult.one().

在 1.4 版本加入.

返回:

The first Row.

Raises:

MultipleResultsFound, NoResultFound

参见

AsyncResult.first()

AsyncResult.one_or_none()

AsyncResult.scalar_one()

method sqlalchemy.ext.asyncio.AsyncResult.async one_or_none() → Row[Unpack[_Ts]] | None

Return at most one result or raise an exception.

Returns None if the result has no rows. Raises MultipleResultsFound if multiple rows are returned.

在 1.4 版本加入.

返回:

The first Row or None if no row is available.

Raises:

MultipleResultsFound

参见

AsyncResult.first()

AsyncResult.one()

method sqlalchemy.ext.asyncio.AsyncResult.async partitions(size: int | None = None) → AsyncIterator[Sequence[Row[Unpack[_Ts]]]]

Iterate through sub-lists of rows of the size given.

An async iterator is returned:

async def scroll_results(connection):
    result = await connection.stream(select(users_table))

    async for partition in result.partitions(100):
        print("list of rows: %s" % partition)

Refer to Result.partitions() in the synchronous SQLAlchemy API for a complete behavioral description.

method sqlalchemy.ext.asyncio.AsyncResult.async scalar() → Any

Fetch the first column of the first row, and close the result set.

Returns None if there are no rows to fetch.

No validation is performed to test if additional rows remain.

After calling this method, the object is fully closed, e.g. the CursorResult.close() method will have been called.

返回:

a Python scalar value, or None if no rows remain.

method sqlalchemy.ext.asyncio.AsyncResult.async scalar_one() → Any

Return exactly one scalar result or raise an exception.

This is equivalent to calling AsyncResult.scalars() and then AsyncScalarResult.one().

参见

AsyncScalarResult.one()

AsyncResult.scalars()

method sqlalchemy.ext.asyncio.AsyncResult.async scalar_one_or_none() → Any | None

Return exactly one scalar result or None.

This is equivalent to calling AsyncResult.scalars() and then AsyncScalarResult.one_or_none().

参见

AsyncScalarResult.one_or_none()

AsyncResult.scalars()

method sqlalchemy.ext.asyncio.AsyncResult.scalars(index: _KeyIndexType = 0) → AsyncScalarResult[Any]

Return an AsyncScalarResult filtering object which will return single elements rather than Row objects.

Refer to Result.scalars() in the synchronous SQLAlchemy API for a complete behavioral description.

参数:

index – integer or row key indicating the column to be fetched from each row, defaults to 0 indicating the first column.

返回:

a new AsyncScalarResult filtering object referring to this AsyncResult object.

attribute sqlalchemy.ext.asyncio.AsyncResult.t

Apply a “typed tuple” typing filter to returned rows.

自 2.1.0 版本弃用: The AsyncResult.t attribute is deprecated, Row now behaves like a tuple and can unpack types directly.

The AsyncResult.t attribute is a synonym for calling the AsyncResult.tuples() method.

在 2.0 版本加入.

参见

Row 现在直接表示单个列类型，无需 Tuple - describes a migration path from this workaround for SQLAlchemy 2.1.

method sqlalchemy.ext.asyncio.AsyncResult.tuples() → AsyncTupleResult[Tuple[Unpack[_Ts]]]

Apply a “typed tuple” typing filter to returned rows.

自 2.1.0 版本弃用: The AsyncResult.tuples() method is deprecated, Row now behaves like a tuple and can unpack types directly.

This method returns the same AsyncResult object at runtime, however annotates as returning a AsyncTupleResult object that will indicate to PEP 484 typing tools that plain typed Tuple instances are returned rather than rows. This allows tuple unpacking and __getitem__ access of Row objects to by typed, for those cases where the statement invoked itself included typing information.

在 2.0 版本加入.

返回:

the AsyncTupleResult type at typing time.

参见

Row 现在直接表示单个列类型，无需 Tuple - describes a migration path from this workaround for SQLAlchemy 2.1.

AsyncResult.t - shorter synonym

Row.t - Row version

method sqlalchemy.ext.asyncio.AsyncResult.unique(strategy: _UniqueFilterType | None = None) → Self

Apply unique filtering to the objects returned by this AsyncResult.

Refer to Result.unique() in the synchronous SQLAlchemy API for a complete behavioral description.

method sqlalchemy.ext.asyncio.AsyncResult.yield_per(num: int) → Self

inherited from the FilterResult.yield_per() method of FilterResult

Configure the row-fetching strategy to fetch num rows at a time.

The FilterResult.yield_per() method is a pass through to the Result.yield_per() method. See that method’s documentation for usage notes.

在 1.4.40 版本加入: - added FilterResult.yield_per() so that the method is available on all result set implementations

参见

使用服务器端游标（又称流结果） - describes Core behavior for Result.yield_per()

使用 Yield Per 获取大量结果集 - in the ORM 查询指南

class sqlalchemy.ext.asyncio.AsyncScalarResult

A wrapper for a AsyncResult that returns scalar values rather than Row values.

The AsyncScalarResult object is acquired by calling the AsyncResult.scalars() method.

Refer to the ScalarResult object in the synchronous SQLAlchemy API for a complete behavioral description.

在 1.4 版本加入.

Members

all(), close(), closed, fetchall(), fetchmany(), first(), one(), one_or_none(), partitions(), unique(), yield_per()

Class signature

class sqlalchemy.ext.asyncio.AsyncScalarResult (sqlalchemy.ext.asyncio.AsyncCommon)

method sqlalchemy.ext.asyncio.AsyncScalarResult.async all() → Sequence[_R]

Return all scalar values in a list.

Equivalent to AsyncResult.all() except that scalar values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncScalarResult.async close() → None

inherited from the AsyncCommon.close() method of AsyncCommon

Close this result.

attribute sqlalchemy.ext.asyncio.AsyncScalarResult.closed

inherited from the AsyncCommon.closed attribute of AsyncCommon

proxies the .closed attribute of the underlying result object, if any, else raises AttributeError.

在 2.0.0b3 版本加入.

method sqlalchemy.ext.asyncio.AsyncScalarResult.async fetchall() → Sequence[_R]

A synonym for the AsyncScalarResult.all() method.

method sqlalchemy.ext.asyncio.AsyncScalarResult.async fetchmany(size: int | None = None) → Sequence[_R]

Fetch many objects.

Equivalent to AsyncResult.fetchmany() except that scalar values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncScalarResult.async first() → _R | None

Fetch the first object or None if no object is present.

Equivalent to AsyncResult.first() except that scalar values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncScalarResult.async one() → _R

Return exactly one object or raise an exception.

Equivalent to AsyncResult.one() except that scalar values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncScalarResult.async one_or_none() → _R | None

Return at most one object or raise an exception.

Equivalent to AsyncResult.one_or_none() except that scalar values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncScalarResult.async partitions(size: int | None = None) → AsyncIterator[Sequence[_R]]

Iterate through sub-lists of elements of the size given.

Equivalent to AsyncResult.partitions() except that scalar values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncScalarResult.unique(strategy: _UniqueFilterType | None = None) → Self

Apply unique filtering to the objects returned by this AsyncScalarResult.

See AsyncResult.unique() for usage details.

method sqlalchemy.ext.asyncio.AsyncScalarResult.yield_per(num: int) → Self

inherited from the FilterResult.yield_per() method of FilterResult

Configure the row-fetching strategy to fetch num rows at a time.

The FilterResult.yield_per() method is a pass through to the Result.yield_per() method. See that method’s documentation for usage notes.

在 1.4.40 版本加入: - added FilterResult.yield_per() so that the method is available on all result set implementations

参见

使用服务器端游标（又称流结果） - describes Core behavior for Result.yield_per()

使用 Yield Per 获取大量结果集 - in the ORM 查询指南

class sqlalchemy.ext.asyncio.AsyncMappingResult

A wrapper for a AsyncResult that returns dictionary values rather than Row values.

The AsyncMappingResult object is acquired by calling the AsyncResult.mappings() method.

Refer to the MappingResult object in the synchronous SQLAlchemy API for a complete behavioral description.

在 1.4 版本加入.

Members

all(), close(), closed, columns(), fetchall(), fetchmany(), fetchone(), first(), keys(), one(), one_or_none(), partitions(), unique(), yield_per()

Class signature

class sqlalchemy.ext.asyncio.AsyncMappingResult (sqlalchemy.engine._WithKeys, sqlalchemy.ext.asyncio.AsyncCommon)

method sqlalchemy.ext.asyncio.AsyncMappingResult.async all() → Sequence[RowMapping]

Return all rows in a list.

Equivalent to AsyncResult.all() except that RowMapping values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncMappingResult.async close() → None

inherited from the AsyncCommon.close() method of AsyncCommon

Close this result.

attribute sqlalchemy.ext.asyncio.AsyncMappingResult.closed

inherited from the AsyncCommon.closed attribute of AsyncCommon

proxies the .closed attribute of the underlying result object, if any, else raises AttributeError.

在 2.0.0b3 版本加入.

method sqlalchemy.ext.asyncio.AsyncMappingResult.columns(*col_expressions: _KeyIndexType) → Self

Establish the columns that should be returned in each row.

method sqlalchemy.ext.asyncio.AsyncMappingResult.async fetchall() → Sequence[RowMapping]

A synonym for the AsyncMappingResult.all() method.

method sqlalchemy.ext.asyncio.AsyncMappingResult.async fetchmany(size: int | None = None) → Sequence[RowMapping]

Fetch many rows.

Equivalent to AsyncResult.fetchmany() except that RowMapping values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncMappingResult.async fetchone() → RowMapping | None

Fetch one object.

Equivalent to AsyncResult.fetchone() except that RowMapping values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncMappingResult.async first() → RowMapping | None

Fetch the first object or None if no object is present.

Equivalent to AsyncResult.first() except that RowMapping values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncMappingResult.keys() → RMKeyView

inherited from the sqlalchemy.engine._WithKeys.keys method of sqlalchemy.engine._WithKeys

Return an iterable view which yields the string keys that would be represented by each Row.

The keys can represent the labels of the columns returned by a core statement or the names of the orm classes returned by an orm execution.

The view also can be tested for key containment using the Python in operator, which will test both for the string keys represented in the view, as well as for alternate keys such as column objects.

在 1.4 版本发生变更: a key view object is returned rather than a plain list.

method sqlalchemy.ext.asyncio.AsyncMappingResult.async one() → RowMapping

Return exactly one object or raise an exception.

Equivalent to AsyncResult.one() except that RowMapping values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncMappingResult.async one_or_none() → RowMapping | None

Return at most one object or raise an exception.

Equivalent to AsyncResult.one_or_none() except that RowMapping values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncMappingResult.async partitions(size: int | None = None) → AsyncIterator[Sequence[RowMapping]]

Iterate through sub-lists of elements of the size given.

Equivalent to AsyncResult.partitions() except that RowMapping values, rather than Row objects, are returned.

method sqlalchemy.ext.asyncio.AsyncMappingResult.unique(strategy: _UniqueFilterType | None = None) → Self

Apply unique filtering to the objects returned by this AsyncMappingResult.

See AsyncResult.unique() for usage details.

method sqlalchemy.ext.asyncio.AsyncMappingResult.yield_per(num: int) → Self

inherited from the FilterResult.yield_per() method of FilterResult

Configure the row-fetching strategy to fetch num rows at a time.

The FilterResult.yield_per() method is a pass through to the Result.yield_per() method. See that method’s documentation for usage notes.

在 1.4.40 版本加入: - added FilterResult.yield_per() so that the method is available on all result set implementations

参见

使用服务器端游标（又称流结果） - describes Core behavior for Result.yield_per()

使用 Yield Per 获取大量结果集 - in the ORM 查询指南

class sqlalchemy.ext.asyncio.AsyncTupleResult

A AsyncResult that’s typed as returning plain Python tuples instead of rows.

Since Row acts like a tuple in every way already, this class is a typing only class, regular AsyncResult is still used at runtime.

Class signature

class sqlalchemy.ext.asyncio.AsyncTupleResult (sqlalchemy.ext.asyncio.AsyncCommon, sqlalchemy.util.langhelpers.TypingOnly)

ORM 会话 API 文档

ORM Session API Documentation

Object Name	Description
async_object_session(instance)	Return the AsyncSession to which the given instance belongs.
async_scoped_session	Provides scoped management of AsyncSession objects.
async_session(session)	Return the AsyncSession which is proxying the given Session object, if any.
async_sessionmaker	A configurable AsyncSession factory.
AsyncAttrs	Mixin class which provides an awaitable accessor for all attributes.
AsyncSession	Asyncio version of Session.
AsyncSessionTransaction	A wrapper for the ORM SessionTransaction object.
close_all_sessions()	Close all AsyncSession sessions.

function sqlalchemy.ext.asyncio.async_object_session(instance: object) → AsyncSession | None

Return the AsyncSession to which the given instance belongs.

This function makes use of the sync-API function object_session to retrieve the Session which refers to the given instance, and from there links it to the original AsyncSession.

If the AsyncSession has been garbage collected, the return value is None.

This functionality is also available from the InstanceState.async_session accessor.

参数:

instance – an ORM mapped instance

返回:

an AsyncSession object, or None.

在 1.4.18 版本加入.

function sqlalchemy.ext.asyncio.async_session(session: Session) → AsyncSession | None

Return the AsyncSession which is proxying the given Session object, if any.

参数:

session – a Session instance.

返回:

a AsyncSession instance, or None.

在 1.4.18 版本加入.

function async sqlalchemy.ext.asyncio.close_all_sessions() → None

Close all AsyncSession sessions.

在 2.0.23 版本加入.

参见

close_all_sessions()

class sqlalchemy.ext.asyncio.async_sessionmaker

A configurable AsyncSession factory.

The async_sessionmaker factory works in the same way as the sessionmaker factory, to generate new AsyncSession objects when called, creating them given the configurational arguments established here.

e.g.:

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker


async def run_some_sql(
    async_session: async_sessionmaker[AsyncSession],
) -> None:
    async with async_session() as session:
        session.add(SomeObject(data="object"))
        session.add(SomeOtherObject(name="other object"))
        await session.commit()


async def main() -> None:
    # an AsyncEngine, which the AsyncSession will use for connection
    # resources
    engine = create_async_engine(
        "postgresql+asyncpg://scott:tiger@localhost/"
    )

    # create a reusable factory for new AsyncSession instances
    async_session = async_sessionmaker(engine)

    await run_some_sql(async_session)

    await engine.dispose()

The async_sessionmaker is useful so that different parts of a program can create new AsyncSession objects with a fixed configuration established up front. Note that AsyncSession objects may also be instantiated directly when not using async_sessionmaker.

在 2.0 版本加入: async_sessionmaker provides a sessionmaker class that’s dedicated to the AsyncSession object, including pep-484 typing support.

参见

概要 - ORM - shows example use

sessionmaker - general overview of the

sessionmaker architecture

打开和关闭会话 - introductory text on creating sessions using sessionmaker.

Members

__call__(), __init__(), begin(), configure()

Class signature

class sqlalchemy.ext.asyncio.async_sessionmaker (typing.Generic)

method sqlalchemy.ext.asyncio.async_sessionmaker.__call__(**local_kw: Any) → _AS

Produce a new AsyncSession object using the configuration established in this async_sessionmaker.

In Python, the __call__ method is invoked on an object when it is “called” in the same way as a function:

AsyncSession = async_sessionmaker(async_engine, expire_on_commit=False)
session = AsyncSession()  # invokes sessionmaker.__call__()

method sqlalchemy.ext.asyncio.async_sessionmaker.__init__(bind: Optional[_AsyncSessionBind] = None, *, class_: Type[_AS] = <class 'sqlalchemy.ext.asyncio.session.AsyncSession'>, autoflush: bool = True, expire_on_commit: bool = True, info: Optional[_InfoType] = None, **kw: Any)

Construct a new async_sessionmaker.

All arguments here except for class_ correspond to arguments accepted by Session directly. See the AsyncSession.__init__() docstring for more details on parameters.

method sqlalchemy.ext.asyncio.async_sessionmaker.begin() → _AsyncSessionContextManager[_AS]

Produce a context manager that both provides a new AsyncSession as well as a transaction that commits.

e.g.:

async def main():
    Session = async_sessionmaker(some_engine)

    async with Session.begin() as session:
        session.add(some_object)

    # commits transaction, closes session

method sqlalchemy.ext.asyncio.async_sessionmaker.configure(**new_kw: Any) → None

(Re)configure the arguments for this async_sessionmaker.

e.g.:

AsyncSession = async_sessionmaker(some_engine)

AsyncSession.configure(bind=create_async_engine("sqlite+aiosqlite://"))

class sqlalchemy.ext.asyncio.async_scoped_session

Provides scoped management of AsyncSession objects.

See the section 使用 asyncio 作用域会话 for usage details.

在 1.4.19 版本加入.

Members

__call__(), __init__(), aclose(), add(), add_all(), autoflush, begin(), begin_nested(), bind, close(), close_all(), commit(), configure(), connection(), delete(), delete_all(), deleted, dirty, execute(), expire(), expire_all(), expunge(), expunge_all(), flush(), get(), get_bind(), get_one(), identity_key(), identity_map, info, invalidate(), is_active, is_modified(), merge(), merge_all(), new, no_autoflush, object_session(), refresh(), remove(), reset(), rollback(), scalar(), scalars(), session_factory, stream(), stream_scalars()

Class signature

class sqlalchemy.ext.asyncio.async_scoped_session (typing.Generic)

method sqlalchemy.ext.asyncio.async_scoped_session.__call__(**kw: Any) → _AS

Return the current AsyncSession, creating it using the scoped_session.session_factory if not present.

参数:

**kw – Keyword arguments will be passed to the scoped_session.session_factory callable, if an existing AsyncSession is not present. If the AsyncSession is present and keyword arguments have been passed, InvalidRequestError is raised.

method sqlalchemy.ext.asyncio.async_scoped_session.__init__(session_factory: async_sessionmaker[_AS], scopefunc: Callable[[], Any])

Construct a new async_scoped_session.

参数:

• session_factory – a factory to create new AsyncSession instances. This is usually, but not necessarily, an instance of async_sessionmaker.

• scopefunc – function which defines the current scope. A function such as asyncio.current_task may be useful here.

method sqlalchemy.ext.asyncio.async_scoped_session.async aclose() → None

A synonym for AsyncSession.close().

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

The AsyncSession.aclose() name is specifically to support the Python standard library @contextlib.aclosing context manager function.

在 2.0.20 版本加入.

method sqlalchemy.ext.asyncio.async_scoped_session.add(instance: object, *, _warn: bool = True) → None

Place an object into this Session.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

Objects that are in the transient state when passed to the Session.add() method will move to the pending state, until the next flush, at which point they will move to the persistent state.

Objects that are in the detached state when passed to the Session.add() method will move to the persistent state directly.

If the transaction used by the Session is rolled back, objects which were transient when they were passed to Session.add() will be moved back to the transient state, and will no longer be present within this Session.

参见

Session.add_all()

添加新项目或现有项目 - at 使用会话的基础知识

method sqlalchemy.ext.asyncio.async_scoped_session.add_all(instances: Iterable[object]) → None

Add the given collection of instances to this Session.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

See the documentation for Session.add() for a general behavioral description.

参见

Session.add()

添加新项目或现有项目 - at 使用会话的基础知识

attribute sqlalchemy.ext.asyncio.async_scoped_session.autoflush

Proxy for the Session.autoflush attribute on behalf of the AsyncSession class.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

method sqlalchemy.ext.asyncio.async_scoped_session.begin() → AsyncSessionTransaction

Return an AsyncSessionTransaction object.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

The underlying Session will perform the “begin” action when the AsyncSessionTransaction object is entered:

async with async_session.begin():
    ...  # ORM transaction is begun

Note that database IO will not normally occur when the session-level transaction is begun, as database transactions begin on an on-demand basis. However, the begin block is async to accommodate for a SessionEvents.after_transaction_create() event hook that may perform IO.

For a general description of ORM begin, see Session.begin().

method sqlalchemy.ext.asyncio.async_scoped_session.begin_nested() → AsyncSessionTransaction

Return an AsyncSessionTransaction object which will begin a “nested” transaction, e.g. SAVEPOINT.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Behavior is the same as that of AsyncSession.begin().

For a general description of ORM begin nested, see Session.begin_nested().

参见

可序列化隔离/保存点/事务 DDL（asyncio版本） - special workarounds required with the SQLite asyncio driver in order for SAVEPOINT to work correctly.

attribute sqlalchemy.ext.asyncio.async_scoped_session.bind

Proxy for the AsyncSession.bind attribute on behalf of the async_scoped_session class.

method sqlalchemy.ext.asyncio.async_scoped_session.async close() → None

Close out the transactional resources and ORM objects used by this AsyncSession.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

参见

Session.close() - main documentation for “close”

关闭 - detail on the semantics of AsyncSession.close() and AsyncSession.reset().

async classmethod sqlalchemy.ext.asyncio.async_scoped_session.close_all() → None

Close all AsyncSession sessions.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

自 2.0 版本弃用: The AsyncSession.close_all() method is deprecated and will be removed in a future release. Please refer to close_all_sessions().

method sqlalchemy.ext.asyncio.async_scoped_session.async commit() → None

Commit the current transaction in progress.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

参见

Session.commit() - main documentation for “commit”

method sqlalchemy.ext.asyncio.async_scoped_session.configure(**kwargs: Any) → None

reconfigure the sessionmaker used by this scoped_session.

See sessionmaker.configure().

method sqlalchemy.ext.asyncio.async_scoped_session.async connection(bind_arguments: _BindArguments | None = None, execution_options: CoreExecuteOptionsParameter | None = None, **kw: Any) → AsyncConnection

Return a AsyncConnection object corresponding to this Session object’s transactional state.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

This method may also be used to establish execution options for the database connection used by the current transaction.

在 1.4.24 版本加入: Added **kw arguments which are passed through to the underlying Session.connection() method.

参见

Session.connection() - main documentation for “connection”

method sqlalchemy.ext.asyncio.async_scoped_session.async delete(instance: object) → None

Mark an instance as deleted.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

The database delete operation occurs upon flush().

As this operation may need to cascade along unloaded relationships, it is awaitable to allow for those queries to take place.

参见

Session.delete() - main documentation for delete

method sqlalchemy.ext.asyncio.async_scoped_session.async delete_all(instances: Iterable[object]) → None

Calls AsyncSession.delete() on multiple instances.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

参见

Session.delete_all() - main documentation for delete_all

attribute sqlalchemy.ext.asyncio.async_scoped_session.deleted

The set of all instances marked as ‘deleted’ within this Session

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

attribute sqlalchemy.ext.asyncio.async_scoped_session.dirty

The set of all persistent instances considered dirty.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

E.g.:

some_mapped_object in session.dirty

Instances are considered dirty when they were modified but not deleted.

Note that this ‘dirty’ calculation is ‘optimistic’; most attribute-setting or collection modification operations will mark an instance as ‘dirty’ and place it in this set, even if there is no net change to the attribute’s value. At flush time, the value of each attribute is compared to its previously saved value, and if there’s no net change, no SQL operation will occur (this is a more expensive operation so it’s only done at flush time).

To check if an instance has actionable net changes to its attributes, use the Session.is_modified() method.

method sqlalchemy.ext.asyncio.async_scoped_session.async execute(statement: Executable, params: _CoreAnyExecuteParams | None = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: _BindArguments | None = None, **kw: Any) → Result[Unpack[TupleAny]]

Execute a statement and return a buffered Result object.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

参见

Session.execute() - main documentation for execute

method sqlalchemy.ext.asyncio.async_scoped_session.expire(instance: object, attribute_names: Iterable[str] | None = None) → None

Expire the attributes on an instance.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

Marks the attributes of an instance as out of date. When an expired attribute is next accessed, a query will be issued to the Session object’s current transactional context in order to load all expired attributes for the given instance. Note that a highly isolated transaction will return the same values as were previously read in that same transaction, regardless of changes in database state outside of that transaction.

To expire all objects in the Session simultaneously, use Session.expire_all().

The Session object’s default behavior is to expire all state whenever the Session.rollback() or Session.commit() methods are called, so that new state can be loaded for the new transaction. For this reason, calling Session.expire() only makes sense for the specific case that a non-ORM SQL statement was emitted in the current transaction.

参数:

• instance – The instance to be refreshed.

• attribute_names – optional list of string attribute names indicating a subset of attributes to be expired.

参见

刷新/过期 - introductory material

Session.expire()

Session.refresh()

Query.populate_existing()

method sqlalchemy.ext.asyncio.async_scoped_session.expire_all() → None

Expires all persistent instances within this Session.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

When any attributes on a persistent instance is next accessed, a query will be issued using the Session object’s current transactional context in order to load all expired attributes for the given instance. Note that a highly isolated transaction will return the same values as were previously read in that same transaction, regardless of changes in database state outside of that transaction.

To expire individual objects and individual attributes on those objects, use Session.expire().

The Session object’s default behavior is to expire all state whenever the Session.rollback() or Session.commit() methods are called, so that new state can be loaded for the new transaction. For this reason, calling Session.expire_all() is not usually needed, assuming the transaction is isolated.

参见

刷新/过期 - introductory material

Session.expire()

Session.refresh()

Query.populate_existing()

method sqlalchemy.ext.asyncio.async_scoped_session.expunge(instance: object) → None

Remove the instance from this Session.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

This will free all internal references to the instance. Cascading will be applied according to the expunge cascade rule.

method sqlalchemy.ext.asyncio.async_scoped_session.expunge_all() → None

Remove all object instances from this Session.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

This is equivalent to calling expunge(obj) on all objects in this Session.

method sqlalchemy.ext.asyncio.async_scoped_session.async flush(objects: Sequence[Any] | None = None) → None

Flush all the object changes to the database.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

参见

Session.flush() - main documentation for flush

method sqlalchemy.ext.asyncio.async_scoped_session.async get(entity: _EntityBindKey[_O], ident: _PKIdentityArgument, *, options: Sequence[ORMOption] | None = None, populate_existing: bool = False, with_for_update: ForUpdateParameter = None, identity_token: Any | None = None, execution_options: OrmExecuteOptionsParameter = {}) → _O | None

Return an instance based on the given primary key identifier, or None if not found.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

参见

Session.get() - main documentation for get

method sqlalchemy.ext.asyncio.async_scoped_session.get_bind(mapper: _EntityBindKey[_O] | None = None, clause: ClauseElement | None = None, bind: _SessionBind | None = None, **kw: Any) → Engine | Connection

Return a “bind” to which the synchronous proxied Session is bound.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Unlike the Session.get_bind() method, this method is currently not used by this AsyncSession in any way in order to resolve engines for requests.

备注

This method proxies directly to the Session.get_bind() method, however is currently not useful as an override target, in contrast to that of the Session.get_bind() method. The example below illustrates how to implement custom Session.get_bind() schemes that work with AsyncSession and AsyncEngine.

The pattern introduced at 自定义垂直分区 illustrates how to apply a custom bind-lookup scheme to a Session given a set of Engine objects. To apply a corresponding Session.get_bind() implementation for use with a AsyncSession and AsyncEngine objects, continue to subclass Session and apply it to AsyncSession using AsyncSession.sync_session_class. The inner method must continue to return Engine instances, which can be acquired from a AsyncEngine using the AsyncEngine.sync_engine attribute:

# using example from "Custom Vertical Partitioning"


import random

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.orm import Session

# construct async engines w/ async drivers
engines = {
    "leader": create_async_engine("sqlite+aiosqlite:///leader.db"),
    "other": create_async_engine("sqlite+aiosqlite:///other.db"),
    "follower1": create_async_engine("sqlite+aiosqlite:///follower1.db"),
    "follower2": create_async_engine("sqlite+aiosqlite:///follower2.db"),
}


class RoutingSession(Session):
    def get_bind(self, mapper=None, clause=None, **kw):
        # within get_bind(), return sync engines
        if mapper and issubclass(mapper.class_, MyOtherClass):
            return engines["other"].sync_engine
        elif self._flushing or isinstance(clause, (Update, Delete)):
            return engines["leader"].sync_engine
        else:
            return engines[
                random.choice(["follower1", "follower2"])
            ].sync_engine


# apply to AsyncSession using sync_session_class
AsyncSessionMaker = async_sessionmaker(sync_session_class=RoutingSession)

The Session.get_bind() method is called in a non-asyncio, implicitly non-blocking context in the same manner as ORM event hooks and functions that are invoked via AsyncSession.run_sync(), so routines that wish to run SQL commands inside of Session.get_bind() can continue to do so using blocking-style code, which will be translated to implicitly async calls at the point of invoking IO on the database drivers.

method sqlalchemy.ext.asyncio.async_scoped_session.async get_one(entity: _EntityBindKey[_O], ident: _PKIdentityArgument, *, options: Sequence[ORMOption] | None = None, populate_existing: bool = False, with_for_update: ForUpdateParameter = None, identity_token: Any | None = None, execution_options: OrmExecuteOptionsParameter = {}) → _O

Return an instance based on the given primary key identifier, or raise an exception if not found.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Raises sqlalchemy.orm.exc.NoResultFound if the query selects no rows.

..versionadded: 2.0.22

参见

Session.get_one() - main documentation for get_one

classmethod sqlalchemy.ext.asyncio.async_scoped_session.identity_key(class_: Type[Any] | None = None, ident: Any | Tuple[Any, ...] = None, *, instance: Any | None = None, row: Row[Unpack[TupleAny]] | RowMapping | None = None, identity_token: Any | None = None) → _IdentityKeyType[Any]

Return an identity key.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

This is an alias of identity_key().

attribute sqlalchemy.ext.asyncio.async_scoped_session.identity_map

Proxy for the Session.identity_map attribute on behalf of the AsyncSession class.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

attribute sqlalchemy.ext.asyncio.async_scoped_session.info

A user-modifiable dictionary.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

The initial value of this dictionary can be populated using the info argument to the Session constructor or sessionmaker constructor or factory methods. The dictionary here is always local to this Session and can be modified independently of all other Session objects.

method sqlalchemy.ext.asyncio.async_scoped_session.async invalidate() → None

Close this Session, using connection invalidation.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

For a complete description, see Session.invalidate().

attribute sqlalchemy.ext.asyncio.async_scoped_session.is_active

True if this Session not in “partial rollback” state.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

在 1.4 版本发生变更: The Session no longer begins a new transaction immediately, so this attribute will be False when the Session is first instantiated.

“partial rollback” state typically indicates that the flush process of the Session has failed, and that the Session.rollback() method must be emitted in order to fully roll back the transaction.

If this Session is not in a transaction at all, the Session will autobegin when it is first used, so in this case Session.is_active will return True.

Otherwise, if this Session is within a transaction, and that transaction has not been rolled back internally, the Session.is_active will also return True.

参见

“由于刷新期间出现先前的异常，此 Session 的事务已被回滚。”（或类似）

Session.in_transaction()

method sqlalchemy.ext.asyncio.async_scoped_session.is_modified(instance: object, include_collections: bool = True) → bool

Return True if the given instance has locally modified attributes.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

This method retrieves the history for each instrumented attribute on the instance and performs a comparison of the current value to its previously flushed or committed value, if any.

It is in effect a more expensive and accurate version of checking for the given instance in the Session.dirty collection; a full test for each attribute’s net “dirty” status is performed.

E.g.:

return session.is_modified(someobject)

A few caveats to this method apply:

• Instances present in the Session.dirty collection may report False when tested with this method. This is because the object may have received change events via attribute mutation, thus placing it in Session.dirty, but ultimately the state is the same as that loaded from the database, resulting in no net change here.

• Scalar attributes may not have recorded the previously set value when a new value was applied, if the attribute was not loaded, or was expired, at the time the new value was received - in these cases, the attribute is assumed to have a change, even if there is ultimately no net change against its database value. SQLAlchemy in most cases does not need the “old” value when a set event occurs, so it skips the expense of a SQL call if the old value isn’t present, based on the assumption that an UPDATE of the scalar value is usually needed, and in those few cases where it isn’t, is less expensive on average than issuing a defensive SELECT.

  The “old” value is fetched unconditionally upon set only if the attribute container has the active_history flag set to True. This flag is set typically for primary key attributes and scalar object references that are not a simple many-to-one. To set this flag for any arbitrary mapped column, use the active_history argument with column_property().

参数:

• instance – mapped instance to be tested for pending changes.

• include_collections – Indicates if multivalued collections should be included in the operation. Setting this to False is a way to detect only local-column based properties (i.e. scalar columns or many-to-one foreign keys) that would result in an UPDATE for this instance upon flush.

method sqlalchemy.ext.asyncio.async_scoped_session.async merge(instance: _O, *, load: bool = True, options: Sequence[ORMOption] | None = None) → _O

Copy the state of a given instance into a corresponding instance within this AsyncSession.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

参见

Session.merge() - main documentation for merge

method sqlalchemy.ext.asyncio.async_scoped_session.async merge_all(instances: Iterable[_O], *, load: bool = True, options: Sequence[ORMOption] | None = None) → Sequence[_O]

Calls AsyncSession.merge() on multiple instances.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

参见

Session.merge_all() - main documentation for merge_all

attribute sqlalchemy.ext.asyncio.async_scoped_session.new

The set of all instances marked as ‘new’ within this Session.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

attribute sqlalchemy.ext.asyncio.async_scoped_session.no_autoflush

Return a context manager that disables autoflush.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

e.g.:

with session.no_autoflush:

    some_object = SomeClass()
    session.add(some_object)
    # won't autoflush
    some_object.related_thing = session.query(SomeRelated).first()

Operations that proceed within the with: block will not be subject to flushes occurring upon query access. This is useful when initializing a series of objects which involve existing database queries, where the uncompleted object should not yet be flushed.

classmethod sqlalchemy.ext.asyncio.async_scoped_session.object_session(instance: object) → Session | None

Return the Session to which an object belongs.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

Proxied for the Session class on behalf of the AsyncSession class.

This is an alias of object_session().

method sqlalchemy.ext.asyncio.async_scoped_session.async refresh(instance: object, attribute_names: Iterable[str] | None = None, with_for_update: ForUpdateParameter = None) → None

Expire and refresh the attributes on the given instance.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

A query will be issued to the database and all attributes will be refreshed with their current database value.

This is the async version of the Session.refresh() method. See that method for a complete description of all options.

参见

Session.refresh() - main documentation for refresh

method sqlalchemy.ext.asyncio.async_scoped_session.async remove() → None

Dispose of the current AsyncSession, if present.

Different from scoped_session’s remove method, this method would use await to wait for the close method of AsyncSession.

method sqlalchemy.ext.asyncio.async_scoped_session.async reset() → None

Close out the transactional resources and ORM objects used by this Session, resetting the session to its initial state.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

在 2.0.22 版本加入.

参见

Session.reset() - main documentation for “reset”

关闭 - detail on the semantics of AsyncSession.close() and AsyncSession.reset().

method sqlalchemy.ext.asyncio.async_scoped_session.async rollback() → None

Rollback the current transaction in progress.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

参见

Session.rollback() - main documentation for “rollback”

method sqlalchemy.ext.asyncio.async_scoped_session.async scalar(statement: Executable, params: _CoreAnyExecuteParams | None = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: _BindArguments | None = None, **kw: Any) → Any

Execute a statement and return a scalar result.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

参见

Session.scalar() - main documentation for scalar

method sqlalchemy.ext.asyncio.async_scoped_session.async scalars(statement: Executable, params: _CoreAnyExecuteParams | None = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: _BindArguments | None = None, **kw: Any) → ScalarResult[Any]

Execute a statement and return scalar results.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

返回:

a ScalarResult object

在 1.4.24 版本加入: Added AsyncSession.scalars()

在 1.4.26 版本加入: Added async_scoped_session.scalars()

参见

Session.scalars() - main documentation for scalars

AsyncSession.stream_scalars() - streaming version

attribute sqlalchemy.ext.asyncio.async_scoped_session.session_factory: async_sessionmaker[_AS]

The session_factory provided to __init__ is stored in this attribute and may be accessed at a later time. This can be useful when a new non-scoped AsyncSession is needed.

method sqlalchemy.ext.asyncio.async_scoped_session.async stream(statement: Executable, params: _CoreAnyExecuteParams | None = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: _BindArguments | None = None, **kw: Any) → AsyncResult[Unpack[TupleAny]]

Execute a statement and return a streaming AsyncResult object.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

method sqlalchemy.ext.asyncio.async_scoped_session.async stream_scalars(statement: Executable, params: _CoreAnyExecuteParams | None = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: _BindArguments | None = None, **kw: Any) → AsyncScalarResult[Any]

Execute a statement and return a stream of scalar results.

Proxied for the AsyncSession class on behalf of the async_scoped_session class.

返回:

an AsyncScalarResult object

在 1.4.24 版本加入.

参见

Session.scalars() - main documentation for scalars

AsyncSession.scalars() - non streaming version

class sqlalchemy.ext.asyncio.AsyncAttrs

Mixin class which provides an awaitable accessor for all attributes.

E.g.:

from __future__ import annotations

from typing import List

from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship


class Base(AsyncAttrs, DeclarativeBase):
    pass


class A(Base):
    __tablename__ = "a"

    id: Mapped[int] = mapped_column(primary_key=True)
    data: Mapped[str]
    bs: Mapped[List[B]] = relationship()


class B(Base):
    __tablename__ = "b"
    id: Mapped[int] = mapped_column(primary_key=True)
    a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))
    data: Mapped[str]

In the above example, the AsyncAttrs mixin is applied to the declarative Base class where it takes effect for all subclasses. This mixin adds a single new attribute AsyncAttrs.awaitable_attrs to all classes, which will yield the value of any attribute as an awaitable. This allows attributes which may be subject to lazy loading or deferred / unexpiry loading to be accessed such that IO can still be emitted:

a1 = (await async_session.scalars(select(A).where(A.id == 5))).one()

# use the lazy loader on ``a1.bs`` via the ``.awaitable_attrs``
# interface, so that it may be awaited
for b1 in await a1.awaitable_attrs.bs:
    print(b1)

The AsyncAttrs.awaitable_attrs performs a call against the attribute that is approximately equivalent to using the AsyncSession.run_sync() method, e.g.:

for b1 in await async_session.run_sync(lambda sess: a1.bs):
    print(b1)

在 2.0.13 版本加入.

Members

awaitable_attrs

参见

使用 AsyncSession 时防止隐式 IO

attribute sqlalchemy.ext.asyncio.AsyncAttrs.awaitable_attrs

provide a namespace of all attributes on this object wrapped as awaitables.

e.g.:

a1 = (await async_session.scalars(select(A).where(A.id == 5))).one()

some_attribute = await a1.awaitable_attrs.some_deferred_attribute
some_collection = await a1.awaitable_attrs.some_collection

class sqlalchemy.ext.asyncio.AsyncSession

Asyncio version of Session.

The AsyncSession is a proxy for a traditional Session instance.

The AsyncSession is not safe for use in concurrent tasks.. See 会话是线程安全的吗？AsyncSession 在并发任务中共享是否安全？ for background.

在 1.4 版本加入.

To use an AsyncSession with custom Session implementations, see the AsyncSession.sync_session_class parameter.

Members

sync_session_class, __init__(), aclose(), add(), add_all(), autoflush, begin(), begin_nested(), close(), close_all(), commit(), connection(), delete(), delete_all(), deleted, dirty, execute(), expire(), expire_all(), expunge(), expunge_all(), flush(), get(), get_bind(), get_nested_transaction(), get_one(), get_transaction(), identity_key(), identity_map, in_nested_transaction(), in_transaction(), info, invalidate(), is_active, is_modified(), merge(), merge_all(), new, no_autoflush, object_session(), refresh(), reset(), rollback(), run_sync(), scalar(), scalars(), stream(), stream_scalars(), sync_session

Class signature

class sqlalchemy.ext.asyncio.AsyncSession (sqlalchemy.ext.asyncio.base.ReversibleProxy)

attribute sqlalchemy.ext.asyncio.AsyncSession.sync_session_class: Type[Session] = <class 'sqlalchemy.orm.session.Session'>

The class or callable that provides the underlying Session instance for a particular AsyncSession.

At the class level, this attribute is the default value for the AsyncSession.sync_session_class parameter. Custom subclasses of AsyncSession can override this.

At the instance level, this attribute indicates the current class or callable that was used to provide the Session instance for this AsyncSession instance.

在 1.4.24 版本加入.

method sqlalchemy.ext.asyncio.AsyncSession.__init__(bind: _AsyncSessionBind | None = None, *, binds: Dict[_SessionBindKey, _AsyncSessionBind] | None = None, sync_session_class: Type[Session] | None = None, **kw: Any)

Construct a new AsyncSession.

All parameters other than sync_session_class are passed to the sync_session_class callable directly to instantiate a new Session. Refer to Session.__init__() for parameter documentation.

参数:

sync_session_class –

A Session subclass or other callable which will be used to construct the Session which will be proxied. This parameter may be used to provide custom Session subclasses. Defaults to the AsyncSession.sync_session_class class-level attribute.

在 1.4.24 版本加入.

method sqlalchemy.ext.asyncio.AsyncSession.async aclose() → None

A synonym for AsyncSession.close().

The AsyncSession.aclose() name is specifically to support the Python standard library @contextlib.aclosing context manager function.

在 2.0.20 版本加入.

method sqlalchemy.ext.asyncio.AsyncSession.add(instance: object, *, _warn: bool = True) → None

Place an object into this Session.

Proxied for the Session class on behalf of the AsyncSession class.

Objects that are in the transient state when passed to the Session.add() method will move to the pending state, until the next flush, at which point they will move to the persistent state.

Objects that are in the detached state when passed to the Session.add() method will move to the persistent state directly.

If the transaction used by the Session is rolled back, objects which were transient when they were passed to Session.add() will be moved back to the transient state, and will no longer be present within this Session.

参见

Session.add_all()

添加新项目或现有项目 - at 使用会话的基础知识

method sqlalchemy.ext.asyncio.AsyncSession.add_all(instances: Iterable[object]) → None

Add the given collection of instances to this Session.

Proxied for the Session class on behalf of the AsyncSession class.

See the documentation for Session.add() for a general behavioral description.

参见

Session.add()

添加新项目或现有项目 - at 使用会话的基础知识

attribute sqlalchemy.ext.asyncio.AsyncSession.autoflush

Proxy for the Session.autoflush attribute on behalf of the AsyncSession class.

method sqlalchemy.ext.asyncio.AsyncSession.begin() → AsyncSessionTransaction

Return an AsyncSessionTransaction object.

The underlying Session will perform the “begin” action when the AsyncSessionTransaction object is entered:

async with async_session.begin():
    ...  # ORM transaction is begun

Note that database IO will not normally occur when the session-level transaction is begun, as database transactions begin on an on-demand basis. However, the begin block is async to accommodate for a SessionEvents.after_transaction_create() event hook that may perform IO.

For a general description of ORM begin, see Session.begin().

method sqlalchemy.ext.asyncio.AsyncSession.begin_nested() → AsyncSessionTransaction

Return an AsyncSessionTransaction object which will begin a “nested” transaction, e.g. SAVEPOINT.

Behavior is the same as that of AsyncSession.begin().

For a general description of ORM begin nested, see Session.begin_nested().

参见

可序列化隔离/保存点/事务 DDL（asyncio版本） - special workarounds required with the SQLite asyncio driver in order for SAVEPOINT to work correctly.

method sqlalchemy.ext.asyncio.AsyncSession.async close() → None

Close out the transactional resources and ORM objects used by this AsyncSession.

参见

Session.close() - main documentation for “close”

关闭 - detail on the semantics of AsyncSession.close() and AsyncSession.reset().

async classmethod sqlalchemy.ext.asyncio.AsyncSession.close_all() → None

Close all AsyncSession sessions.

自 2.0 版本弃用: The AsyncSession.close_all() method is deprecated and will be removed in a future release. Please refer to close_all_sessions().

method sqlalchemy.ext.asyncio.AsyncSession.async commit() → None

Commit the current transaction in progress.

参见

Session.commit() - main documentation for “commit”

method sqlalchemy.ext.asyncio.AsyncSession.async connection(bind_arguments: _BindArguments | None = None, execution_options: CoreExecuteOptionsParameter | None = None, **kw: Any) → AsyncConnection

Return a AsyncConnection object corresponding to this Session object’s transactional state.

This method may also be used to establish execution options for the database connection used by the current transaction.

在 1.4.24 版本加入: Added **kw arguments which are passed through to the underlying Session.connection() method.

参见

Session.connection() - main documentation for “connection”

method sqlalchemy.ext.asyncio.AsyncSession.async delete(instance: object) → None

Mark an instance as deleted.

The database delete operation occurs upon flush().

As this operation may need to cascade along unloaded relationships, it is awaitable to allow for those queries to take place.

参见

Session.delete() - main documentation for delete

method sqlalchemy.ext.asyncio.AsyncSession.async delete_all(instances: Iterable[object]) → None

Calls AsyncSession.delete() on multiple instances.

参见

Session.delete_all() - main documentation for delete_all

attribute sqlalchemy.ext.asyncio.AsyncSession.deleted

The set of all instances marked as ‘deleted’ within this Session

Proxied for the Session class on behalf of the AsyncSession class.

attribute sqlalchemy.ext.asyncio.AsyncSession.dirty

The set of all persistent instances considered dirty.

Proxied for the Session class on behalf of the AsyncSession class.

E.g.:

some_mapped_object in session.dirty

Instances are considered dirty when they were modified but not deleted.

Note that this ‘dirty’ calculation is ‘optimistic’; most attribute-setting or collection modification operations will mark an instance as ‘dirty’ and place it in this set, even if there is no net change to the attribute’s value. At flush time, the value of each attribute is compared to its previously saved value, and if there’s no net change, no SQL operation will occur (this is a more expensive operation so it’s only done at flush time).

To check if an instance has actionable net changes to its attributes, use the Session.is_modified() method.

method sqlalchemy.ext.asyncio.AsyncSession.async execute(statement: Executable, params: _CoreAnyExecuteParams | None = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: _BindArguments | None = None, **kw: Any) → Result[Unpack[TupleAny]]

Execute a statement and return a buffered Result object.

参见

Session.execute() - main documentation for execute

method sqlalchemy.ext.asyncio.AsyncSession.expire(instance: object, attribute_names: Iterable[str] | None = None) → None

Expire the attributes on an instance.

Proxied for the Session class on behalf of the AsyncSession class.

Marks the attributes of an instance as out of date. When an expired attribute is next accessed, a query will be issued to the Session object’s current transactional context in order to load all expired attributes for the given instance. Note that a highly isolated transaction will return the same values as were previously read in that same transaction, regardless of changes in database state outside of that transaction.

To expire all objects in the Session simultaneously, use Session.expire_all().

The Session object’s default behavior is to expire all state whenever the Session.rollback() or Session.commit() methods are called, so that new state can be loaded for the new transaction. For this reason, calling Session.expire() only makes sense for the specific case that a non-ORM SQL statement was emitted in the current transaction.

参数:

• instance – The instance to be refreshed.

• attribute_names – optional list of string attribute names indicating a subset of attributes to be expired.

参见

刷新/过期 - introductory material

Session.expire()

Session.refresh()

Query.populate_existing()

method sqlalchemy.ext.asyncio.AsyncSession.expire_all() → None

Expires all persistent instances within this Session.

Proxied for the Session class on behalf of the AsyncSession class.

When any attributes on a persistent instance is next accessed, a query will be issued using the Session object’s current transactional context in order to load all expired attributes for the given instance. Note that a highly isolated transaction will return the same values as were previously read in that same transaction, regardless of changes in database state outside of that transaction.

To expire individual objects and individual attributes on those objects, use Session.expire().

The Session object’s default behavior is to expire all state whenever the Session.rollback() or Session.commit() methods are called, so that new state can be loaded for the new transaction. For this reason, calling Session.expire_all() is not usually needed, assuming the transaction is isolated.

参见

刷新/过期 - introductory material

Session.expire()

Session.refresh()

Query.populate_existing()

method sqlalchemy.ext.asyncio.AsyncSession.expunge(instance: object) → None

Remove the instance from this Session.

Proxied for the Session class on behalf of the AsyncSession class.

This will free all internal references to the instance. Cascading will be applied according to the expunge cascade rule.

method sqlalchemy.ext.asyncio.AsyncSession.expunge_all() → None

Remove all object instances from this Session.

Proxied for the Session class on behalf of the AsyncSession class.

This is equivalent to calling expunge(obj) on all objects in this Session.

method sqlalchemy.ext.asyncio.AsyncSession.async flush(objects: Sequence[Any] | None = None) → None

Flush all the object changes to the database.

参见

Session.flush() - main documentation for flush

method sqlalchemy.ext.asyncio.AsyncSession.async get(entity: _EntityBindKey[_O], ident: _PKIdentityArgument, *, options: Sequence[ORMOption] | None = None, populate_existing: bool = False, with_for_update: ForUpdateParameter = None, identity_token: Any | None = None, execution_options: OrmExecuteOptionsParameter = {}) → _O | None

Return an instance based on the given primary key identifier, or None if not found.

参见

Session.get() - main documentation for get

method sqlalchemy.ext.asyncio.AsyncSession.get_bind(mapper: _EntityBindKey[_O] | None = None, clause: ClauseElement | None = None, bind: _SessionBind | None = None, **kw: Any) → Engine | Connection

Return a “bind” to which the synchronous proxied Session is bound.

Unlike the Session.get_bind() method, this method is currently not used by this AsyncSession in any way in order to resolve engines for requests.

备注

This method proxies directly to the Session.get_bind() method, however is currently not useful as an override target, in contrast to that of the Session.get_bind() method. The example below illustrates how to implement custom Session.get_bind() schemes that work with AsyncSession and AsyncEngine.

The pattern introduced at 自定义垂直分区 illustrates how to apply a custom bind-lookup scheme to a Session given a set of Engine objects. To apply a corresponding Session.get_bind() implementation for use with a AsyncSession and AsyncEngine objects, continue to subclass Session and apply it to AsyncSession using AsyncSession.sync_session_class. The inner method must continue to return Engine instances, which can be acquired from a AsyncEngine using the AsyncEngine.sync_engine attribute:

# using example from "Custom Vertical Partitioning"


import random

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.orm import Session

# construct async engines w/ async drivers
engines = {
    "leader": create_async_engine("sqlite+aiosqlite:///leader.db"),
    "other": create_async_engine("sqlite+aiosqlite:///other.db"),
    "follower1": create_async_engine("sqlite+aiosqlite:///follower1.db"),
    "follower2": create_async_engine("sqlite+aiosqlite:///follower2.db"),
}


class RoutingSession(Session):
    def get_bind(self, mapper=None, clause=None, **kw):
        # within get_bind(), return sync engines
        if mapper and issubclass(mapper.class_, MyOtherClass):
            return engines["other"].sync_engine
        elif self._flushing or isinstance(clause, (Update, Delete)):
            return engines["leader"].sync_engine
        else:
            return engines[
                random.choice(["follower1", "follower2"])
            ].sync_engine


# apply to AsyncSession using sync_session_class
AsyncSessionMaker = async_sessionmaker(sync_session_class=RoutingSession)

The Session.get_bind() method is called in a non-asyncio, implicitly non-blocking context in the same manner as ORM event hooks and functions that are invoked via AsyncSession.run_sync(), so routines that wish to run SQL commands inside of Session.get_bind() can continue to do so using blocking-style code, which will be translated to implicitly async calls at the point of invoking IO on the database drivers.

method sqlalchemy.ext.asyncio.AsyncSession.get_nested_transaction() → AsyncSessionTransaction | None

Return the current nested transaction in progress, if any.

返回:

an AsyncSessionTransaction object, or None.

在 1.4.18 版本加入.

method sqlalchemy.ext.asyncio.AsyncSession.async get_one(entity: _EntityBindKey[_O], ident: _PKIdentityArgument, *, options: Sequence[ORMOption] | None = None, populate_existing: bool = False, with_for_update: ForUpdateParameter = None, identity_token: Any | None = None, execution_options: OrmExecuteOptionsParameter = {}) → _O

Return an instance based on the given primary key identifier, or raise an exception if not found.

Raises sqlalchemy.orm.exc.NoResultFound if the query selects no rows.

..versionadded: 2.0.22

参见

Session.get_one() - main documentation for get_one

method sqlalchemy.ext.asyncio.AsyncSession.get_transaction() → AsyncSessionTransaction | None

Return the current root transaction in progress, if any.

返回:

an AsyncSessionTransaction object, or None.

在 1.4.18 版本加入.

classmethod sqlalchemy.ext.asyncio.AsyncSession.identity_key(class_: Type[Any] | None = None, ident: Any | Tuple[Any, ...] = None, *, instance: Any | None = None, row: Row[Unpack[TupleAny]] | RowMapping | None = None, identity_token: Any | None = None) → _IdentityKeyType[Any]

Return an identity key.

Proxied for the Session class on behalf of the AsyncSession class.

This is an alias of identity_key().

attribute sqlalchemy.ext.asyncio.AsyncSession.identity_map

Proxy for the Session.identity_map attribute on behalf of the AsyncSession class.

method sqlalchemy.ext.asyncio.AsyncSession.in_nested_transaction() → bool

Return True if this Session has begun a nested transaction, e.g. SAVEPOINT.

Proxied for the Session class on behalf of the AsyncSession class.

在 1.4 版本加入.

method sqlalchemy.ext.asyncio.AsyncSession.in_transaction() → bool

Return True if this Session has begun a transaction.

Proxied for the Session class on behalf of the AsyncSession class.

在 1.4 版本加入.

参见

Session.is_active

attribute sqlalchemy.ext.asyncio.AsyncSession.info

A user-modifiable dictionary.

Proxied for the Session class on behalf of the AsyncSession class.

The initial value of this dictionary can be populated using the info argument to the Session constructor or sessionmaker constructor or factory methods. The dictionary here is always local to this Session and can be modified independently of all other Session objects.

method sqlalchemy.ext.asyncio.AsyncSession.async invalidate() → None

Close this Session, using connection invalidation.

For a complete description, see Session.invalidate().

attribute sqlalchemy.ext.asyncio.AsyncSession.is_active

True if this Session not in “partial rollback” state.

Proxied for the Session class on behalf of the AsyncSession class.

在 1.4 版本发生变更: The Session no longer begins a new transaction immediately, so this attribute will be False when the Session is first instantiated.

“partial rollback” state typically indicates that the flush process of the Session has failed, and that the Session.rollback() method must be emitted in order to fully roll back the transaction.

If this Session is not in a transaction at all, the Session will autobegin when it is first used, so in this case Session.is_active will return True.

Otherwise, if this Session is within a transaction, and that transaction has not been rolled back internally, the Session.is_active will also return True.

参见

“由于刷新期间出现先前的异常，此 Session 的事务已被回滚。”（或类似）

Session.in_transaction()

method sqlalchemy.ext.asyncio.AsyncSession.is_modified(instance: object, include_collections: bool = True) → bool

Return True if the given instance has locally modified attributes.

Proxied for the Session class on behalf of the AsyncSession class.

This method retrieves the history for each instrumented attribute on the instance and performs a comparison of the current value to its previously flushed or committed value, if any.

It is in effect a more expensive and accurate version of checking for the given instance in the Session.dirty collection; a full test for each attribute’s net “dirty” status is performed.

E.g.:

return session.is_modified(someobject)

A few caveats to this method apply:

• Instances present in the Session.dirty collection may report False when tested with this method. This is because the object may have received change events via attribute mutation, thus placing it in Session.dirty, but ultimately the state is the same as that loaded from the database, resulting in no net change here.

• Scalar attributes may not have recorded the previously set value when a new value was applied, if the attribute was not loaded, or was expired, at the time the new value was received - in these cases, the attribute is assumed to have a change, even if there is ultimately no net change against its database value. SQLAlchemy in most cases does not need the “old” value when a set event occurs, so it skips the expense of a SQL call if the old value isn’t present, based on the assumption that an UPDATE of the scalar value is usually needed, and in those few cases where it isn’t, is less expensive on average than issuing a defensive SELECT.

  The “old” value is fetched unconditionally upon set only if the attribute container has the active_history flag set to True. This flag is set typically for primary key attributes and scalar object references that are not a simple many-to-one. To set this flag for any arbitrary mapped column, use the active_history argument with column_property().

参数:

• instance – mapped instance to be tested for pending changes.

• include_collections – Indicates if multivalued collections should be included in the operation. Setting this to False is a way to detect only local-column based properties (i.e. scalar columns or many-to-one foreign keys) that would result in an UPDATE for this instance upon flush.

method sqlalchemy.ext.asyncio.AsyncSession.async merge(instance: _O, *, load: bool = True, options: Sequence[ORMOption] | None = None) → _O

Copy the state of a given instance into a corresponding instance within this AsyncSession.

参见

Session.merge() - main documentation for merge

method sqlalchemy.ext.asyncio.AsyncSession.async merge_all(instances: Iterable[_O], *, load: bool = True, options: Sequence[ORMOption] | None = None) → Sequence[_O]

Calls AsyncSession.merge() on multiple instances.

参见

Session.merge_all() - main documentation for merge_all

attribute sqlalchemy.ext.asyncio.AsyncSession.new

The set of all instances marked as ‘new’ within this Session.

Proxied for the Session class on behalf of the AsyncSession class.

attribute sqlalchemy.ext.asyncio.AsyncSession.no_autoflush

Return a context manager that disables autoflush.

Proxied for the Session class on behalf of the AsyncSession class.

e.g.:

with session.no_autoflush:

    some_object = SomeClass()
    session.add(some_object)
    # won't autoflush
    some_object.related_thing = session.query(SomeRelated).first()

Operations that proceed within the with: block will not be subject to flushes occurring upon query access. This is useful when initializing a series of objects which involve existing database queries, where the uncompleted object should not yet be flushed.

classmethod sqlalchemy.ext.asyncio.AsyncSession.object_session(instance: object) → Session | None

Return the Session to which an object belongs.

Proxied for the Session class on behalf of the AsyncSession class.

This is an alias of object_session().

method sqlalchemy.ext.asyncio.AsyncSession.async refresh(instance: object, attribute_names: Iterable[str] | None = None, with_for_update: ForUpdateParameter = None) → None

Expire and refresh the attributes on the given instance.

A query will be issued to the database and all attributes will be refreshed with their current database value.

This is the async version of the Session.refresh() method. See that method for a complete description of all options.

参见

Session.refresh() - main documentation for refresh

method sqlalchemy.ext.asyncio.AsyncSession.async reset() → None

Close out the transactional resources and ORM objects used by this Session, resetting the session to its initial state.

在 2.0.22 版本加入.

参见

Session.reset() - main documentation for “reset”

关闭 - detail on the semantics of AsyncSession.close() and AsyncSession.reset().

method sqlalchemy.ext.asyncio.AsyncSession.async rollback() → None

Rollback the current transaction in progress.

参见

Session.rollback() - main documentation for “rollback”

method sqlalchemy.ext.asyncio.AsyncSession.async run_sync(fn: ~typing.Callable[[~typing.Concatenate[~sqlalchemy.orm.session.Session, ~_P]], ~sqlalchemy.ext.asyncio.session._T], *arg: ~typing.~_P, **kw: ~typing.~_P) → _T

Invoke the given synchronous (i.e. not async) callable, passing a synchronous-style Session as the first argument.

This method allows traditional synchronous SQLAlchemy functions to run within the context of an asyncio application.

E.g.:

def some_business_method(session: Session, param: str) -> str:
    """A synchronous function that does not require awaiting

    :param session: a SQLAlchemy Session, used synchronously

    :return: an optional return value is supported

    """
    session.add(MyObject(param=param))
    session.flush()
    return "success"


async def do_something_async(async_engine: AsyncEngine) -> None:
    """an async function that uses awaiting"""

    with AsyncSession(async_engine) as async_session:
        # run some_business_method() with a sync-style
        # Session, proxied into an awaitable
        return_code = await async_session.run_sync(
            some_business_method, param="param1"
        )
        print(return_code)

This method maintains the asyncio event loop all the way through to the database connection by running the given callable in a specially instrumented greenlet.

小技巧

The provided callable is invoked inline within the asyncio event loop, and will block on traditional IO calls. IO within this callable should only call into SQLAlchemy’s asyncio database APIs which will be properly adapted to the greenlet context.

参见

AsyncAttrs - a mixin for ORM mapped classes that provides a similar feature more succinctly on a per-attribute basis

AsyncConnection.run_sync()

在 asyncio 下运行同步方法和函数

method sqlalchemy.ext.asyncio.AsyncSession.async scalar(statement: Executable, params: _CoreAnyExecuteParams | None = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: _BindArguments | None = None, **kw: Any) → Any

Execute a statement and return a scalar result.

参见

Session.scalar() - main documentation for scalar

method sqlalchemy.ext.asyncio.AsyncSession.async scalars(statement: Executable, params: _CoreAnyExecuteParams | None = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: _BindArguments | None = None, **kw: Any) → ScalarResult[Any]

Execute a statement and return scalar results.

返回:

a ScalarResult object

在 1.4.24 版本加入: Added AsyncSession.scalars()

在 1.4.26 版本加入: Added async_scoped_session.scalars()

参见

Session.scalars() - main documentation for scalars

AsyncSession.stream_scalars() - streaming version

method sqlalchemy.ext.asyncio.AsyncSession.async stream(statement: Executable, params: _CoreAnyExecuteParams | None = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: _BindArguments | None = None, **kw: Any) → AsyncResult[Unpack[TupleAny]]

Execute a statement and return a streaming AsyncResult object.

method sqlalchemy.ext.asyncio.AsyncSession.async stream_scalars(statement: Executable, params: _CoreAnyExecuteParams | None = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: _BindArguments | None = None, **kw: Any) → AsyncScalarResult[Any]

Execute a statement and return a stream of scalar results.

返回:

an AsyncScalarResult object

在 1.4.24 版本加入.

参见

Session.scalars() - main documentation for scalars

AsyncSession.scalars() - non streaming version

attribute sqlalchemy.ext.asyncio.AsyncSession.sync_session: Session

Reference to the underlying Session this AsyncSession proxies requests towards.

This instance can be used as an event target.

参见

使用 asyncio 扩展的事件

class sqlalchemy.ext.asyncio.AsyncSessionTransaction

A wrapper for the ORM SessionTransaction object.

This object is provided so that a transaction-holding object for the AsyncSession.begin() may be returned.

The object supports both explicit calls to AsyncSessionTransaction.commit() and AsyncSessionTransaction.rollback(), as well as use as an async context manager.

在 1.4 版本加入.

Members

commit(), rollback()

Class signature

class sqlalchemy.ext.asyncio.AsyncSessionTransaction (sqlalchemy.ext.asyncio.base.ReversibleProxy, sqlalchemy.ext.asyncio.base.StartableContext)

method sqlalchemy.ext.asyncio.AsyncSessionTransaction.async commit() → None

Commit this AsyncTransaction.

method sqlalchemy.ext.asyncio.AsyncSessionTransaction.async rollback() → None

Roll back this AsyncTransaction.