Asynchronous I/O (asyncio)

Python asyncioのサポート。asyncio互換のダイアレクトを使用した、CoreおよびORMの使用のサポートが含まれます。

New in version 1.4.

Warning

Apple M1 Architecture を含む多くのプラットフォーム用の重要なプラットフォームインストールノートについては、 Asyncio Platform Installation Notes (Including Apple M1) をお読みください。

See also

Asynchronous IO Support for Core and ORM - 最初の機能の紹介

Asyncio Integration - asyncio拡張内でのCoreとORMの使用例を示すサンプルスクリプト。

Asyncio Platform Installation Notes (Including Apple M1)

asyncio拡張モジュールはPython 3のみを必要とします。また、 greenlet ライブラリにも依存します。この依存関係は、次のような一般的なマシンプラットフォームにデフォルトでインストールされます。

x86_64 aarch64 ppc64le amd64 win32

上記のプラ ットフォームでは、 greenlet が事前に構築されたwheelファイルを提供することが知られています。他のプラットフォームでは、 greenletはデフォルトではインストールされません ; greenletの現在のファイルリストは Greenlet - Download Files で見ることができます。 Apple M1 を含め、多くのアーキテクチャが省略されていることに注意してください。

SQLAlchemyをインストールする際に、使用しているプラットフォームに関わらず greenlet の依存関係が存在することを確認するには、 [asyncio] etuptools extra を次のようにインストールします。これには、 pipgreenlet をインストールするよう指示することも含まれます。

pip install sqlalchemy[asyncio]

事前に構築されたwheelファイルを持たないプラットフォームに greenlet をインストールすることは、 greenlet がソースから構築されることを意味し、Pythonの開発ライブラリも存在する必要があることに注意してください。

Synopsis - Core

コアとして使用する場合、 create_async_engine() 関数は AsyncEngine のインスタンスを作成します。このインスタンスは従来の Engine APIの非同期バージョンを提供します。 AsyncEngineAsyncEngine.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 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

上記の AsyncConnection.run_sync() メソッドは、awaitableフックは含まれていません。 MetaData.create_all() のような特別なDDL関数を呼び出すために使用できます。

Tip

AsyncEngine オブジェクトをコンテキスト外に出てガベージコレクションされるスコープで使用する場合は、上記の例の async_main 関数に示されているように、 await を使用して AsyncEngine.dispose() メソッドを呼び出すことをお勧めします。これにより、接続プールによって開かれているすべての接続が、待機可能なコンテキスト内で適切に処理されることが保証されます。ブロッキングIOを使用する場合とは異なり、SQLAlchemyは、 await を呼び出す機会がないため、 __del__weakref finalizers などのメソッド内でこれらの接続を適切に処理することができません。エンジンがスコープ外になったときに明示的に処理しないと、ガベージコレクション内で RuntimeError:Event loop is closed という形式に似た警告が標準出力に出力される可能性があります。

AsyncConnection は、 AsyncResult オブジェクトを返す AsyncConnection.stream() メソッドによる”ストリーミング”APIも備えています。この結果オブジェクトはサーバ側のカーソルを使用し、非同期イテレータのような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,))

Synopsis - ORM

2.0 style クエリを使用すると、 AsyncSession クラスは完全なORM機能を提供します。

デフォルトの使用モードでは、 lazy loading や、ORM関係や列属性を含むその他の期限切れ属性へのアクセスを避けるために、特別な注意を払う必要があります。次のセクション Preventing Implicit IO when Using AsyncSession で詳しく説明します。

Warning

:class`_asyncio.AsyncSession` の単一インスタンスは、複数の並行タスクで使用するには 安全ではありません 。背景については Using AsyncSession with Concurrent TasksIs the Session thread-safe? Is AsyncSession safe to share in concurrent tasks? の節を参照してください。

次の例は、マッパーとセッション構成を含む完全な例を示しています。:

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

上の例では、 AsyncSession はオプションの async_sessionmaker ヘルパーを使ってインスタンス化されます。このヘルパーは、新しい AsyncSession オブジェクトのファクトリを、固定されたパラメータのセットとともに提供します。ここでは、特定のデータベースURLに対して AsyncEngine と関連付けることも含まれています。その後、他のメソッドに渡され、Pythonの非同期コンテキストマネージャ(つまり、 async with: 文)で使用され、ブロックの最後で自動的に閉じられます。これは AsyncSession.close() メソッドを呼び出すのと同じです。

Using AsyncSession with Concurrent Tasks

AsyncSession オブジェクトは、 変更可能でステートフルなオブジェクト で、 処理中の単一のステートフルなデータベーストランザクション を表します。例えば、 asyncio.gather() のようなAPIを持つasyncioで並行タスクを使用する場合、個々のタスクごとに 別々の AsyncSession を使用する必要があります

SessionAsyncSession の並行ワークロードでの使用方法に関する一般的な説明については、 Is the Session thread-safe? Is AsyncSession safe to share in concurrent tasks? の節を参照してください。

Preventing Implicit IO when Using AsyncSession

従来のasyncioを使用すると、アプリケーションはIO-on-attributeアクセスが発生する可能性のあるポイントを避ける必要があります。これを支援するために使用できるテクニックを以下に示します。その多くは前の例で説明されています。

  • 遅延読み込み関係、遅延列または式である属性、または有効期限シナリオでアクセスされている属性は、 AsyncAttrs mixinを利用できます。このmixinは、特定のクラス、より一般的には宣言型の Base スーパークラスに追加されると、任意の属性をawaitableとして提供するアクセサー AsyncAttrs.awaitable_attrs を提供します:

    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 ...

    eager loadingが使用されていない時に、新しくロードされた`A`のインスタンスの`A.bs`コレクションにアクセスするには、通常 lazy loading を使用します。これは成功するために通常データベースにIOを発行しますが、暗黙的なIOが許可されていないためasyncioの下では失敗します。事前にロード操作を行わずにasyncioの下でこの属性に直接アクセスするには、 AsyncAttrs.awaitable_attrs プレフィックスを指定することで、awaitableとして属性にアクセスできます:

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

    AsyncAttrs ミックスインは、 AsyncSession.run_sync() メソッドでも使用される内部的なアプローチに簡潔なファサードを提供します。

    New in version 2.0.13.

    See also

    AsyncAttrs

  • コレクションは、SQLAlchemy 2.0の Write Only Relationships 機能を使用して、暗黙的にIOを発行しない 書き込み専用コレクション に置き換えることができます。この機能を使用すると、コレクションは決して読み取られず、明示的なSQL呼び出しを使用してクエリされるだけです。asyncioで使用される書き込み専用コレクションの例については、 Asyncio Integration セクションの例 async_orm_writeonly.py を参照してください。

    書き込み専用コレクションを使用する場合、プログラムの動作は単純で、コレクションに関して簡単に予測できます。ただし、これらのコレクションの多くを一度にロードするための組み込みシステムがないという欠点があり、その場合は手動で実行する必要があります。そのため、以下の箇条書きの多くは、従来の遅延ロードされた関係をasyncioで使用する場合の特定のテクニックを扱っていますが、これにはより注意が必要です。

  • AsyncAttrs を使わない場合は、デフォルトでSQLを出力しないように、関係を lazy="raise" で宣言することができます。コレクションをロードするには、代わりに eager loading を使います。

  • もっとも役に立つeager loading戦略は selectinload() eager loaderです。これは前の例で、 await session.execute() コールのスコープ内で A.bs コレクションをeager loadするために使われています:

    stmt = select(A).options(selectinload(A.bs))
  • 新しいオブジェクトを構築する場合、 コレクションには常にデフォルトの空のコレクション (上記の例のリストなど)が割り当てられます:

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

    これにより、上記の A オブジェクトの .bs コレクションは、 A オブジェクトがフラッシュされたときに存在し、読み取り可能になります。そうでなければ、 A がフラッシュされたときに .bs がアンロードされ、アクセス時にエラーが発生します。

  • AsyncSession は、Falseに設定された Session.expire_on_commit を使って設定されます。これにより、 AsyncSession.commit() の呼び出しの後に、属性にアクセスする最後の行のように、オブジェクトの属性にアクセスできます:

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

その他のガイドラインは次のとおりです。:

  • lazy-loaded関係**は AsyncSession.refresh() を使って asyncio の下に明示的にロードすることができます。 もし 必要な属性名が Session.refresh.attribute_names に明示的に渡されるなら、例えば:

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

    もちろん、遅延ロードを必要とせずにコレクションをセットアップできるように、事前にEager Loadingを使用することをお勧めします。

    New in version 2.0.4: Session.refresh.attribute_names パラメータで明示的に名前が付けられている場合に、遅延ロードされた関係を強制的にロードするための AsyncSession.refresh() および基礎となる Session.refresh() メソッドのサポートが追加されました。以前のバージョンでは、パラメータで名前が付けられていても、関係は自動的にスキップされていました。

  • Cascades に記載されている all カスケードオプションの使用は避け、必要なカスケード機能を明示的に列挙してください。 all カスケードオプションは、特に refresh-expire 設定を意味します。これは、 AsyncSession.refresh() メソッドが関連するオブジェクトの属性を期限切れにすることを意味しますが、 relationship() 内でEager Loadingが設定されていないと仮定すると、必ずしもそれらの関連するオブジェクトを更新せず、期限切れの状態のままにします。

  • Dynamic Relationship Loaders で説明されている”動的”関係ローダー戦略は、デフォルトではasyncioアプローチと互換性がありません。 Running Synchronous Methods and Functions under 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と完全に互換性があり、推奨されます。

    See also

    “Dynamic” relationship loaders superseded by “Write Only” - 2.0スタイルへの移行に関する注意

  • MySQL 8など、RETURNINGをサポートしていないデータベースでasyncioを使用する場合、 Mapper.eager_defaults オプションを使用しない限り、生成されたタイムスタンプなどのサーバのデフォルト値は、新しくフラッシュされたオブジェクトでは使用できません。SQLAlchemy 2.0では、この動作は、行が挿入されたときにRETURNINGを使用して新しい値を取得するPostgreSQL、SQLite、MariaDBなどのバックエンドに自動的に適用されます。

Running Synchronous Methods and Functions under asyncio

Deep Alchemy

このアプローチは基本的に、SQLAlchemyが最初にasyncioインターフェースを提供できるメカニズムを公開しています。そうすることに技術的な問題はありませんが、全体的には、このアプローチはおそらく「議論の余地がある」と考えることができます。なぜなら、このアプローチは、asyncioプログラミングモデルの中心的な哲学のいくつかに反しているからです。それは、本質的に、IOが呼び出される可能性のあるプログラミングステートメントは、プログラムがIOが発生する可能性のあるすべての行を明示的にクリアしないように、await コールを 持たなければならない というものです。このアプローチは、その一般的な考え方を変えるものではありません。ただし、一連の同期IO命令を関数呼び出しの範囲内でこの規則から免除し、基本的に単一のawaitableにバンドルすることを可能にします。

従来のSQLAlchemyの「遅延読み込み」をasyncioイベントループ内に統合する代替手段として、 AsyncSession.run_sync() として知られる オプションの メソッドが提供されています。このメソッドは、グリーンレット内で任意のPython関数を実行します。グリーンレットでは、従来の同期プログラミングの概念が、データベースドライバに到達したときに await を使用するように変換されます。ここでの仮説的なアプローチは、asyncio指向のアプリケーションが、 AsyncSession.run_sync() を使用して呼び出される関数にデータベース関連のメソッドをパッケージ化できることです。

上の例を変更して、 A.bs コレクションに selectinload() を使わなければ、これらの属性へのアクセスを別の関数で処理することができます:

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

“sync”ランナ内で特定の関数を実行する上記のアプローチは、 gevent などのイベントベースのプログラミングライブラリの上でSQLAlchemyアプリケーションを実行するアプリケーションと似ています。違いは次のとおりです。

  1. gevent を使用する場合とは異なり、 gevent イベントループに統合しなくても、標準のPython asyncioイベントループまたは任意のカスタムイベントループを引き続き使用できます。

  1. “モンキーパッチ”はまったくありません。上の例では実際のasyncioドライバを使用しており、基礎となるSQLAlchemy接続プールも接続をプールするためにPythonの組み込みの asyncio.Queue を使用しています。

  1. プログラムは、非同期/待機コードと、同期コードを使用する含まれる関数を、実質的にパフォーマンスを低下させることなく自由に切り替えることができます。”スレッド・エクゼキュータ”や追加の待機または同期は使用されていません。

  1. 基盤となるネットワークドライバも純粋なPython asyncioの概念を使用しており、 geventeventlet が提供するようなサードパーティのネットワークライブラリは使用されていません。

Using events with the asyncio extension

SQLAlchemy event system は、asyncio拡張によって直接公開されません。つまり、SQLAlchemyイベントハンドラの”async”バージョンはまだ存在しません。

ただし、asyncio拡張は通常の同期SQLAlchemy APIを取り囲むため、通常の”同期”スタイルのイベントハンドラは、asyncioが使用されなかった場合と同様に自由に使用できる。

以下に詳述するように、asyncio-facing APIでイベントを登録するための現在の戦略は2つあります。:

  • 同じ型のすべてのインスタンス(例えば、すべての AsyncSession インスタンス)をターゲットにして、クラスレベルでイベントを登録するには、対応するsync-styleクラスを使用してください。例えば、 AsyncSession クラスに対して SessionEvents.before_commit() イベントを登録するには、ターゲットとして Session クラスを使用してください。

asyncioコンテキスト内にあるイベントハンドラ内で動作する場合、 Connection のようなオブジェクトは、通常の”同期”の方法で動作し続け、awaitasync の使用を必要としません。メッセージが最終的にasyncioデータベースアダプタで受信されると、呼び出しスタイルは透過的にasyncio呼び出しスタイルに戻されます。 PoolEvents.connect() のようなDBAPIレベルの接続を渡されるイベントの場合、オブジェクトは pep-249 準拠の”接続”オブジェクトで、syncスタイルの呼び出しをasyncioドライバに適応させます。

Examples of Event Listeners with Async Engines / Sessions / Sessionmakers ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

非同期API構成体に関連付けられたsyncスタイルのイベントハンドラの例を以下に示します。 AsyncEngineのコアイベント:

この例では、 ConnectionEventsPoolEvents のターゲットとして、 AsyncEngineAsyncEngine.sync_engine 属性にアクセスします:

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!
  • AsyncSessionのORMイベント

    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!
  • async_sessionmakerのORMイベント

    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の内部で使用できるように適応させます。このようなイベントハンドラのユーザ定義実装が、最終的な”ドライバ”接続を直接使用し、そのドライバ接続上で待機可能なメソッドのみを使用する必要がある場合に、特別なユースケースが発生します。そのような例の1つは、asyncpgドライバによって提供される .set_type_codec() メソッドです。

このユースケースに対応するために、SQLAlchemyの AdaptedConnection クラスにはメソッド AdaptedConnection.run_async() が用意されています。このメソッドは、イベントハンドラやその他のSQLAlchemy内部の”同期”コンテキスト内で待機可能な関数を呼び出すことができます。このメソッドは、同期スタイルのメソッドを非同期で実行できる AsyncConnection.run_sync() メソッドと直接似ています。

AdaptedConnection.run_async() には、最も内側の”driver”接続を単一の引数として受け入れ、 AdaptedConnection.run_async() メソッドによって呼び出されるawaitableを返す関数を渡す必要があります。指定された関数自体を 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 のインスタンスです。これは、基礎となる非同期のみのドライバレベルの接続オブジェクトへのDB APIのようなインタフェースを提供します。 AdaptedConnection.run_async() メソッドは、基礎となるドライバレベルの接続が動作する待機可能な環境へのアクセスを提供します。

New in version 1.4.30.

Using multiple asyncio event loops

複数のイベントループを使用するアプリケーション、例えばasyncioとマルチスレッドを組み合わせたような珍しいケースでは、デフォルトのプール実装を使用する場合、同じ AsyncEngine を異なるイベントループと共有してはいけません。

AsyncEngine があるイベントループから別のイベントループに渡された場合、新しいイベントループで再利用する前にメソッド AsyncEngine.dispose() を呼び出す必要があります。そうしないと、 Task <Task pending ...> got Future attached to a different loop のような RuntimeError が発生する可能性があります。

同じエンジンを異なるループ間で共有する必要がある場合は、 NullPool を使用してプーリングを無効にし、エンジンが接続を2回以上使用しないように設定する必要があります:

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

Using asyncio scoped session

scoped_session オブジェクトを持つスレッド化されたSQLAlchemyで使用される”scoped session”パターンは、 async_scoped_session と呼ばれる適応バージョンを使用して、asyncioでも使用できます。

Tip

SQLAlchemyは一般的に、新しい開発のための”スコープ付き”パターンを推奨していません。これは、スレッドまたはタスク内の作業が完了したときに明示的に破棄する必要がある可変グローバル状態に依存しているためです。特にasyncioを使用する場合は、 :class`_asyncio.AsyncSession` を必要とする待機可能な関数に直接渡す方がよいでしょう。

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

Warning

async_scoped_session で使用される”scopefunc”は、タスク内で 任意の回数 呼び出されます。これは、基礎となる AsyncSession がアクセスされるたびに1回行われます。したがって、この関数は べき等 かつ軽量であるべきであり、コールバックの確立など、いかなる状態も作成または変更しようとすべきではありません。

Warning

キー”に対して current_task() を使用するには、 async_scoped_session.remove() メソッドを最も外側のawaitable内から呼び出して、タスクが完了したときにキーがレジストリから削除されるようにする必要があります。そうしないと、タスクハンドルと AsyncSession がメモリ内に残り、本質的にメモリリークが発生します。 async_scoped_session.remove() の正しい使用方法を示す次の例を参照してください。

async_scoped_session には scoped_session と同様の**プロキシ動作**が含まれています。つまり、 AsyncSession として直接扱うことができます。ただし、 async_scoped_session.remove() メソッドを含め、通常の await キーワードが必要であることに注意してください。:

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

New in version 1.4.19.

Using the Inspector to inspect schema objects

SQLAlchemyは、( Fine Grained Reflection with Inspector で導入された) Inspector のasyncioバージョンをまだ提供していませんが、 AsyncConnectionAsyncConnection.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())

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.

New in version 1.4.

Parameters:

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.

New in version 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().

New in version 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.

New in version 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")

New in version 1.4.

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.

New in version 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.

Parameters:

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.

See also

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.

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.

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.

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().

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

New in version 1.4.

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.

New in version 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.

See also

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[Any]

Executes a SQL statement construct and return a buffered Result.

Parameters:
  • object

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

  • 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().

Returns:

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.

New in version 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.

New in version 1.4.0b2.

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

Return True if a transaction is in progress.

New in version 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.

New in version 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)

Note

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.

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.

Returns:

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.

Returns:

a ScalarResult object.

New in version 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[Any]]

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.

New in version 2.0.0b3: added context manager support

Returns:

an awaitable object that will yield an AsyncResult object.

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.

New in version 2.0.0b3: added context manager support

Returns:

an awaitable object that will yield an AsyncScalarResult object.

New in version 1.4.24.

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.

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.

class sqlalchemy.ext.asyncio.AsyncTransaction

An asyncio proxy for a Transaction.

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.

Result Set API Documentation

AsyncResult オブジェクトは Result オブジェクトの非同期版です。 AsyncConnection.stream() または AsyncSession.stream() メソッドを使用した場合にのみ返されます。これらのメソッドは、アクティブなデータベースカーソルの上にある結果オブジェクトを返します。

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.

Note

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.

New in version 1.4.

Class signature

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

method sqlalchemy.ext.asyncio.AsyncResult.async all() Sequence[Row[_TP]]

Return all rows in a list.

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

Returns:

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.

New in version 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[_TP]]

A synonym for the AsyncResult.all() method.

New in version 2.0.

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

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.

Returns:

a list of Row objects.

method sqlalchemy.ext.asyncio.AsyncResult.async fetchone() Row[_TP] | 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.

Returns:

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

method sqlalchemy.ext.asyncio.AsyncResult.async first() Row[_TP] | None

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

Closes the result set and discards remaining rows.

Note

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.

Returns:

a Row object, or None if no rows remain.

method sqlalchemy.ext.asyncio.AsyncResult.async freeze() FrozenResult[_TP]

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.

See also

Re-Executing Statements - 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.

Changed in version 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.

Returns:

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

method sqlalchemy.ext.asyncio.AsyncResult.async one() Row[_TP]

Return exactly one row or raise an exception.

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

Note

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().

New in version 1.4.

Returns:

The first Row.

Raises:

MultipleResultsFound, NoResultFound

method sqlalchemy.ext.asyncio.AsyncResult.async one_or_none() Row[_TP] | 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.

New in version 1.4.

Returns:

The first Row or None if no row is available.

Raises:

MultipleResultsFound

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

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.

Returns:

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 AsyncResult.one().

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 AsyncResult.one_or_none().

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.

Parameters:

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

Returns:

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.

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

New in version 2.0.

method sqlalchemy.ext.asyncio.AsyncResult.tuples() AsyncTupleResult[_TP]

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

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.

New in version 2.0.

Returns:

the AsyncTupleResult type at typing time.

See also

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.

New in version 1.4.40: - added FilterResult.yield_per() so that the method is available on all result set implementations

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.

New in version 1.4.

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.

New in version 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.

New in version 1.4.40: - added FilterResult.yield_per() so that the method is available on all result set implementations

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.

New in version 1.4.

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.

New in version 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.

Changed in version 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.

New in version 1.4.40: - added FilterResult.yield_per() so that the method is available on all result set implementations

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 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.

Parameters:

instance – an ORM mapped instance

Returns:

an AsyncSession object, or None.

New in version 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.

Parameters:

session – a Session instance.

Returns:

a AsyncSession instance, or None.

New in version 1.4.18.

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

Close all AsyncSession sessions.

New in version 2.0.23.

See also

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.

New in version 2.0: async_sessionmaker provides a sessionmaker class that’s dedicated to the AsyncSession object, including pep-484 typing support.

See also

Synopsis - ORM - shows example use

sessionmaker - general overview of the

sessionmaker architecture

Opening and Closing a Session - introductory text on creating sessions using sessionmaker.

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 Using asyncio scoped session for usage details.

New in version 1.4.19.

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.

Parameters:

**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.

Parameters:
  • 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.

New in version 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.

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.

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().

See also

Serializable isolation / Savepoints / Transactional DDL (asyncio version) - 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.

See also

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

Closing - 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.

Deprecated since version 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.

See also

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.

New in version 1.4.24: Added **kw arguments which are passed through to the underlying Session.connection() method.

See also

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.

See also

Session.delete() - main documentation for delete

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[Any]

Execute a statement and return a buffered Result object.

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

See also

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.

Parameters:
  • instance – The instance to be refreshed.

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

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.

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.

See also

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.

See also

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.

Note

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 Custom Vertical Partitioning 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

See also

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[Any] | 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.

Changed in version 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.

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().

Parameters:
  • 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.

See also

Session.merge() - main documentation for merge

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.

See also

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.

New in version 2.0.22.

See also

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

Closing - 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.

See also

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.

See also

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.

Returns:

a ScalarResult object

New in version 1.4.24: Added AsyncSession.scalars()

New in version 1.4.26: Added async_scoped_session.scalars()

See also

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[Any]

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.

Returns:

an AsyncScalarResult object

New in version 1.4.24.

See also

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)

New in version 2.0.13.

Members

awaitable_attrs

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 Is the Session thread-safe? Is AsyncSession safe to share in concurrent tasks? for background.

New in version 1.4.

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

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.

New in version 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.

Parameters:

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.

New in version 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.

New in version 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.

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.

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().

See also

Serializable isolation / Savepoints / Transactional DDL (asyncio version) - 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.

See also

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

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

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

Close all AsyncSession sessions.

Deprecated since version 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.

See also

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.

New in version 1.4.24: Added **kw arguments which are passed through to the underlying Session.connection() method.

See also

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.

See also

Session.delete() - main documentation for delete

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[Any]

Execute a statement and return a buffered Result object.

See also

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.

Parameters:
  • instance – The instance to be refreshed.

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

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.

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.

See also

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.

See also

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.

Note

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 Custom Vertical Partitioning 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.

Returns:

an AsyncSessionTransaction object, or None.

New in version 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

See also

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.

Returns:

an AsyncSessionTransaction object, or None.

New in version 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[Any] | 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.

New in version 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.

New in version 1.4.

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.

Changed in version 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.

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().

Parameters:
  • 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.

See also

Session.merge() - main documentation for merge

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.

See also

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.

New in version 2.0.22.

See also

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

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

method sqlalchemy.ext.asyncio.AsyncSession.async rollback() None

Rollback the current transaction in progress.

See also

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.

Tip

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.

See also

AsyncAttrs - a mixin for ORM mapped classes that provides a similar feature more succinctly on a per-attribute basis

AsyncConnection.run_sync()

Running Synchronous Methods and Functions under 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.

See also

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.

Returns:

a ScalarResult object

New in version 1.4.24: Added AsyncSession.scalars()

New in version 1.4.26: Added async_scoped_session.scalars()

See also

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[Any]

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.

Returns:

an AsyncScalarResult object

New in version 1.4.24.

See also

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.

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.

New in version 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.