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 を次のようにインストールします。これには、 pip
に greenlet
をインストールするよう指示することも含まれます。
pip install sqlalchemy[asyncio]
事前に構築されたwheelファイルを持たないプラットフォームに greenlet
をインストールすることは、 greenlet
がソースから構築されることを意味し、Pythonの開発ライブラリも存在する必要があることに注意してください。
Synopsis - Core¶
コアとして使用する場合、 create_async_engine()
関数は AsyncEngine
のインスタンスを作成します。このインスタンスは従来の Engine
APIの非同期バージョンを提供します。 AsyncEngine
は AsyncEngine.connect()
および AsyncEngine.begin()
メソッドを介して AsyncConnection
を提供します。これらのメソッドはどちらも非同期コンテキストマネージャを提供します。 AsyncConnection
は、 AsyncConnection.execute()
メソッドを使用してバッファー化された Result
を提供するか、 AsyncConnection.stream()
メソッドを使用してストリーミングサーバー側の AsyncResult
を提供することで、文を呼び出すことができます。:
>>> import asyncio
>>> from sqlalchemy import Column
>>> from sqlalchemy import MetaData
>>> from sqlalchemy import select
>>> from sqlalchemy import String
>>> from sqlalchemy import Table
>>> from sqlalchemy.ext.asyncio import create_async_engine
>>> meta = MetaData()
>>> t1 = Table("t1", meta, Column("name", String(50), primary_key=True))
>>> async def async_main() -> None:
... engine = create_async_engine("sqlite+aiosqlite://", echo=True)
...
... async with engine.begin() as conn:
... await conn.run_sync(meta.drop_all)
... await conn.run_sync(meta.create_all)
...
... await conn.execute(
... t1.insert(), [{"name": "some name 1"}, {"name": "some name 2"}]
... )
...
... async with engine.connect() as conn:
... # select 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 Tasks と Is 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
を使用する必要があります。
Session
と AsyncSession
の並行ワークロードでの使用方法に関する一般的な説明については、 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
コレクションは、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)
その他のガイドラインは次のとおりです。:
AsyncSession.expire()
のようなメソッドは、AsyncSession.refresh()
を優先して避けるべきです; もし expirationが絶対に必要な場合。asyncioを使用する場合、Session.expire_on_commit
は通常False
に設定されるべきなので、一般的に有効期限は 必要ありません 。
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が設定されていないと仮定すると、必ずしもそれらの関連するオブジェクトを更新せず、期限切れの状態のままにします。
適切なローダオプションを
deferred()
列に使用すべきです。もし使用するのであれば、上で述べたrelationship()
構文のオプションに加えて使用してください。遅延列ロードの背景については Limiting which Columns Load with Column Deferral を参照してください。
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アプリケーションを実行するアプリケーションと似ています。違いは次のとおりです。
gevent
を使用する場合とは異なり、gevent
イベントループに統合しなくても、標準のPython asyncioイベントループまたは任意のカスタムイベントループを引き続き使用できます。
“モンキーパッチ”はまったくありません。上の例では実際のasyncioドライバを使用しており、基礎となるSQLAlchemy接続プールも接続をプールするためにPythonの組み込みの
asyncio.Queue
を使用しています。
プログラムは、非同期/待機コードと、同期コードを使用する含まれる関数を、実質的にパフォーマンスを低下させることなく自由に切り替えることができます。”スレッド・エクゼキュータ”や追加の待機または同期は使用されていません。
基盤となるネットワークドライバも純粋なPython asyncioの概念を使用しており、
gevent
とeventlet
が提供するようなサードパーティのネットワークライブラリは使用されていません。
Using events with the asyncio extension¶
SQLAlchemy event system は、asyncio拡張によって直接公開されません。つまり、SQLAlchemyイベントハンドラの”async”バージョンはまだ存在しません。
ただし、asyncio拡張は通常の同期SQLAlchemy APIを取り囲むため、通常の”同期”スタイルのイベントハンドラは、asyncioが使用されなかった場合と同様に自由に使用できる。
以下に詳述するように、asyncio-facing APIでイベントを登録するための現在の戦略は2つあります。:
イベントは、プロキシされたオブジェクトを参照する
sync
属性にイベントを関連付けることによって、インスタンスレベル(例えば、特定のAsyncEngine
インスタンス)で登録できます。例えば、AsyncEngine
インスタンスに対してPoolEvents.connect()
イベントを登録するには、そのAsyncEngine.sync_engine
属性をターゲットとして使用します。ターゲットには次のものがあります:
同じ型のすべてのインスタンス(例えば、すべての
AsyncSession
インスタンス)をターゲットにして、クラスレベルでイベントを登録するには、対応するsync-styleクラスを使用してください。例えば、AsyncSession
クラスに対してSessionEvents.before_commit()
イベントを登録するには、ターゲットとしてSession
クラスを使用してください。
sessionmaker
レベルで登録するには、async_sessionmaker.sync_session_class
を使って、明示的なsessionmaker
とasync_sessionmaker
を組み合わせ、イベントをsessionmaker
に関連付けます。
asyncioコンテキスト内にあるイベントハンドラ内で動作する場合、 Connection
のようなオブジェクトは、通常の”同期”の方法で動作し続け、await
や async
の使用を必要としません。メッセージが最終的にasyncioデータベースアダプタで受信されると、呼び出しスタイルは透過的にasyncio呼び出しスタイルに戻されます。 PoolEvents.connect()
のようなDBAPIレベルの接続を渡されるイベントの場合、オブジェクトは pep-249 準拠の”接続”オブジェクトで、syncスタイルの呼び出しをasyncioドライバに適応させます。
Examples of Event Listeners with Async Engines / Sessions / Sessionmakers ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
非同期API構成体に関連付けられたsyncスタイルのイベントハンドラの例を以下に示します。 AsyncEngineのコアイベント:
この例では、
ConnectionEvents
とPoolEvents
のターゲットとして、AsyncEngine
のAsyncEngine.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 forSessionEvents
: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 theasync_sessionmaker
using theasync_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バージョンをまだ提供していませんが、 AsyncConnection
の AsyncConnection.run_sync()
メソッドを利用することで、既存のインターフェースをasyncioコンテキストで使用することができます:
import asyncio
from sqlalchemy import inspect
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost/test")
def use_inspector(conn):
inspector = inspect(conn)
# use the inspector
print(inspector.get_view_names())
# return any value to the caller
return inspector.get_table_names()
async def async_main():
async with engine.connect() as conn:
tables = await conn.run_sync(use_inspector)
asyncio.run(async_main())
Engine API Documentation¶
Object Name | Description |
---|---|
async_engine_from_config(configuration[, prefix], **kwargs) |
Create a new AsyncEngine instance using a configuration dictionary. |
An asyncio proxy for a |
|
An asyncio proxy for a |
|
An asyncio proxy for a |
|
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 thecreate_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 thecreate_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 ofengine_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 thecreate_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 thecreate_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.
Members
begin(), clear_compiled_cache(), connect(), dialect, dispose(), driver, echo, engine, execution_options(), get_execution_options(), name, pool, raw_connection(), sync_engine, update_execution_options(), url
Class signature
class
sqlalchemy.ext.asyncio.AsyncEngine
(sqlalchemy.ext.asyncio.base.ProxyComparable
,sqlalchemy.ext.asyncio.AsyncConnectable
)-
method
sqlalchemy.ext.asyncio.AsyncEngine.
begin() AsyncIterator[AsyncConnection] ¶ Return a context manager which when entered will deliver an
AsyncConnection
with anAsyncTransaction
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 theAsyncEngine
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 theConnection.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 itsAsyncConnection.start()
method.
-
attribute
sqlalchemy.ext.asyncio.AsyncEngine.
dialect¶ Proxy for the
Engine.dialect
attribute on behalf of theAsyncEngine
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 thisEngine
, so when they are closed individually, eventually thePool
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
-
attribute
sqlalchemy.ext.asyncio.AsyncEngine.
driver¶ Driver name of the
Dialect
in use by thisEngine
.Proxied for the
Engine
class on behalf of theAsyncEngine
class.
-
attribute
sqlalchemy.ext.asyncio.AsyncEngine.
echo¶ When
True
, enable log output for this element.Proxied for the
Engine
class on behalf of theAsyncEngine
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 loglevellogging.INFO
will be set for the logger, whereas the string valuedebug
will set the loglevel tologging.DEBUG
.
-
attribute
sqlalchemy.ext.asyncio.AsyncEngine.
engine¶ Returns this
Engine
.Proxied for the
Engine
class on behalf of theAsyncEngine
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 provideAsyncConnection
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 theAsyncEngine
class.See also
-
attribute
sqlalchemy.ext.asyncio.AsyncEngine.
name¶ String name of the
Dialect
in use by thisEngine
.Proxied for the
Engine
class on behalf of theAsyncEngine
class.
-
attribute
sqlalchemy.ext.asyncio.AsyncEngine.
pool¶ Proxy for the
Engine.pool
attribute on behalf of theAsyncEngine
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
thisAsyncEngine
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 theAsyncEngine
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 tocreate_engine()
.
-
attribute
sqlalchemy.ext.asyncio.AsyncEngine.
url¶ Proxy for the
Engine.url
attribute on behalf of theAsyncEngine
class.
-
method
- class sqlalchemy.ext.asyncio.AsyncConnection¶
An asyncio proxy for a
Connection
.AsyncConnection
is acquired using theAsyncEngine.connect()
method ofAsyncEngine
: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.
Members
aclose(), begin(), begin_nested(), close(), closed, commit(), connection, default_isolation_level, dialect, exec_driver_sql(), execute(), execution_options(), get_nested_transaction(), get_raw_connection(), get_transaction(), in_nested_transaction(), in_transaction(), info, invalidate(), invalidated, rollback(), run_sync(), scalar(), scalars(), start(), stream(), stream_scalars(), sync_connection, sync_engine
Class signature
class
sqlalchemy.ext.asyncio.AsyncConnection
(sqlalchemy.ext.asyncio.base.ProxyComparable
,sqlalchemy.ext.asyncio.base.StartableContext
,sqlalchemy.ext.asyncio.AsyncConnectable
)-
method
sqlalchemy.ext.asyncio.AsyncConnection.
async aclose() None ¶ A synonym for
AsyncConnection.close()
.The
AsyncConnection.aclose()
name is specifically to support the Python standard library@contextlib.aclosing
context manager function.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 theAsyncConnection
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 theConnection.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 theAsyncConnection
class.This value is independent of the
Connection.execution_options.isolation_level
andEngine.execution_options.isolation_level
execution options, and is determined by theDialect
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 levelcreate_engine.isolation_level
- set perEngine
isolation levelConnection.execution_options.isolation_level
- set perConnection
isolation level
-
attribute
sqlalchemy.ext.asyncio.AsyncConnection.
dialect¶ Proxy for the
Connection.dialect
attribute on behalf of theAsyncConnection
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
andExecutable
hierarchies, including:DDL
and objects which inherit fromExecutableDDLElement
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 DBAPIcursor.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 currentTransaction
, which is then proxied in a newAsyncTransaction
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 anAdaptedConnection
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 currentTransaction
, which is then proxied in a newAsyncTransaction
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 underlyingConnection
.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 theAsyncConnection.closed
attribute isTrue
, then accessing this attribute will raiseResourceClosedError
.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 theAsyncConnection
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 theConnection.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 asMetaData.create_all()
, given anAsyncConnection
that needs to be provided toMetaData.create_all()
as aConnection
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 theConnection.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 theConnection.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 Pythonwith:
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 theAsyncResult
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 anasync with fn():
orawait fn()
style.New in version 2.0.0b3: added context manager support
- Returns:
an awaitable object that will yield an
AsyncResult
object.
See also
-
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 theConnection.stream()
method. Parameters are equivalent.The
AsyncConnection.stream_scalars()
method supports optional context manager use against theAsyncScalarResult
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 anasync with fn():
orawait 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.
See also
-
attribute
sqlalchemy.ext.asyncio.AsyncConnection.
sync_connection: Connection | None¶ Reference to the sync-style
Connection
thisAsyncConnection
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
thisAsyncConnection
is associated with via its underlyingConnection
.This instance can be used as an event target.
-
method
- class sqlalchemy.ext.asyncio.AsyncTransaction¶
An asyncio proxy for a
Transaction
.Members
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 Pythonwith:
block.
-
method
Result Set API Documentation¶
AsyncResult
オブジェクトは Result
オブジェクトの非同期版です。 AsyncConnection.stream()
または AsyncSession.stream()
メソッドを使用した場合にのみ返されます。これらのメソッドは、アクティブなデータベースカーソルの上にある結果オブジェクトを返します。
Object Name | Description |
---|---|
A wrapper for a |
|
An asyncio wrapper around a |
|
A wrapper for a |
|
A |
- 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 theAsyncConnection.stream()
andAsyncSession.stream()
methods.Note
As is the case with
Result
, this object is used for ORM results returned byAsyncSession.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 legacyQuery
object. For in-Python de-duplication of instances or rows, use theAsyncResult.unique()
modifier method.New in version 1.4.
Members
all(), close(), closed, columns(), fetchall(), fetchmany(), fetchone(), first(), freeze(), keys(), mappings(), one(), one_or_none(), partitions(), scalar(), scalar_one(), scalar_one_or_none(), scalars(), t, tuples(), unique(), yield_per()
Class signature
class
sqlalchemy.ext.asyncio.AsyncResult
(sqlalchemy.engine._WithKeys
,sqlalchemy.ext.asyncio.AsyncCommon
)-
method
sqlalchemy.ext.asyncio.AsyncResult.
async all() Sequence[Row[_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 ofAsyncCommon
Close this result.
-
attribute
sqlalchemy.ext.asyncio.AsyncResult.
closed¶ inherited from the
AsyncCommon.closed
attribute ofAsyncCommon
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.
See also
-
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 theAsyncResult
object directly.- Returns:
a
Row
object if no filters are applied, orNone
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 combineAsyncResult.scalars()
andAsyncResult.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 thisAsyncResult
; 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.See also
- 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 newResult
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 ofsqlalchemy.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 ofRow
objects.- Returns:
a new
AsyncMappingResult
filtering object referring to the underlyingResult
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, orMultipleResultsFound
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 combineAsyncResult.scalars()
andAsyncResult.one()
.New in version 1.4.
- Returns:
The first
Row
.- Raises:
-
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. RaisesMultipleResultsFound
if multiple rows are returned.New in version 1.4.
- Returns:
The first
Row
orNone
if no row is available.- Raises:
-
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 thenAsyncResult.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 thenAsyncResult.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 thanRow
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 thisAsyncResult
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 theAsyncResult.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 aAsyncTupleResult
object that will indicate to PEP 484 typing tools that plain typedTuple
instances are returned rather than rows. This allows tuple unpacking and__getitem__
access ofRow
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.
-
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 ofFilterResult
Configure the row-fetching strategy to fetch
num
rows at a time.The
FilterResult.yield_per()
method is a pass through to theResult.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 implementationsSee also
Using Server Side Cursors (a.k.a. stream results) - describes Core behavior for
Result.yield_per()
Fetching Large Result Sets with Yield Per - in the ORM Querying Guide
-
method
- class sqlalchemy.ext.asyncio.AsyncScalarResult¶
A wrapper for a
AsyncResult
that returns scalar values rather thanRow
values.The
AsyncScalarResult
object is acquired by calling theAsyncResult.scalars()
method.Refer to the
ScalarResult
object in the synchronous SQLAlchemy API for a complete behavioral description.New in version 1.4.
Members
all(), close(), closed, fetchall(), fetchmany(), first(), one(), one_or_none(), partitions(), unique(), yield_per()
Class signature
class
sqlalchemy.ext.asyncio.AsyncScalarResult
(sqlalchemy.ext.asyncio.AsyncCommon
)-
method
sqlalchemy.ext.asyncio.AsyncScalarResult.
async all() Sequence[_R] ¶ Return all scalar values in a list.
Equivalent to
AsyncResult.all()
except that scalar values, rather thanRow
objects, are returned.
-
method
sqlalchemy.ext.asyncio.AsyncScalarResult.
async close() None ¶ inherited from the
AsyncCommon.close()
method ofAsyncCommon
Close this result.
-
attribute
sqlalchemy.ext.asyncio.AsyncScalarResult.
closed¶ inherited from the
AsyncCommon.closed
attribute ofAsyncCommon
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 thanRow
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 thanRow
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 thanRow
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 thanRow
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 thanRow
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 ofFilterResult
Configure the row-fetching strategy to fetch
num
rows at a time.The
FilterResult.yield_per()
method is a pass through to theResult.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 implementationsSee also
Using Server Side Cursors (a.k.a. stream results) - describes Core behavior for
Result.yield_per()
Fetching Large Result Sets with Yield Per - in the ORM Querying Guide
-
method
- class sqlalchemy.ext.asyncio.AsyncMappingResult¶
A wrapper for a
AsyncResult
that returns dictionary values rather thanRow
values.The
AsyncMappingResult
object is acquired by calling theAsyncResult.mappings()
method.Refer to the
MappingResult
object in the synchronous SQLAlchemy API for a complete behavioral description.New in version 1.4.
Members
all(), close(), closed, columns(), fetchall(), fetchmany(), fetchone(), first(), keys(), one(), one_or_none(), partitions(), unique(), yield_per()
Class signature
class
sqlalchemy.ext.asyncio.AsyncMappingResult
(sqlalchemy.engine._WithKeys
,sqlalchemy.ext.asyncio.AsyncCommon
)-
method
sqlalchemy.ext.asyncio.AsyncMappingResult.
async all() Sequence[RowMapping] ¶ Return all rows in a list.
Equivalent to
AsyncResult.all()
except thatRowMapping
values, rather thanRow
objects, are returned.
-
method
sqlalchemy.ext.asyncio.AsyncMappingResult.
async close() None ¶ inherited from the
AsyncCommon.close()
method ofAsyncCommon
Close this result.
-
attribute
sqlalchemy.ext.asyncio.AsyncMappingResult.
closed¶ inherited from the
AsyncCommon.closed
attribute ofAsyncCommon
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 thatRowMapping
values, rather thanRow
objects, are returned.
-
method
sqlalchemy.ext.asyncio.AsyncMappingResult.
async fetchone() RowMapping | None ¶ Fetch one object.
Equivalent to
AsyncResult.fetchone()
except thatRowMapping
values, rather thanRow
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 thatRowMapping
values, rather thanRow
objects, are returned.
-
method
sqlalchemy.ext.asyncio.AsyncMappingResult.
keys() RMKeyView ¶ inherited from the
sqlalchemy.engine._WithKeys.keys
method ofsqlalchemy.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 thatRowMapping
values, rather thanRow
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 thatRowMapping
values, rather thanRow
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 thatRowMapping
values, rather thanRow
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 ofFilterResult
Configure the row-fetching strategy to fetch
num
rows at a time.The
FilterResult.yield_per()
method is a pass through to theResult.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 implementationsSee also
Using Server Side Cursors (a.k.a. stream results) - describes Core behavior for
Result.yield_per()
Fetching Large Result Sets with Yield Per - in the ORM Querying Guide
-
method
- 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, regularAsyncResult
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 |
Provides scoped management of |
|
async_session(session) |
Return the |
A configurable |
|
Mixin class which provides an awaitable accessor for all attributes. |
|
Asyncio version of |
|
A wrapper for the ORM |
|
Close all |
- 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 theSession
which refers to the given instance, and from there links it to the originalAsyncSession
.If the
AsyncSession
has been garbage collected, the return value isNone
.This functionality is also available from the
InstanceState.async_session
accessor.- Parameters:
instance¶ – an ORM mapped instance
- Returns:
an
AsyncSession
object, orNone
.
New in version 1.4.18.
- function sqlalchemy.ext.asyncio.async_session(session: Session) AsyncSession | None ¶
Return the
AsyncSession
which is proxying the givenSession
object, if any.- Parameters:
- Returns:
a
AsyncSession
instance, orNone
.
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 thesessionmaker
factory, to generate newAsyncSession
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 newAsyncSession
objects with a fixed configuration established up front. Note thatAsyncSession
objects may also be instantiated directly when not usingasync_sessionmaker
.New in version 2.0:
async_sessionmaker
provides asessionmaker
class that’s dedicated to theAsyncSession
object, including pep-484 typing support.See also
Synopsis - ORM - shows example use
sessionmaker
- general overview of thesessionmaker
architecture
Opening and Closing a Session - introductory text on creating sessions using
sessionmaker
.Members
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 thisasync_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 bySession
directly. See theAsyncSession.__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.
Members
__call__(), __init__(), aclose(), add(), add_all(), autoflush, begin(), begin_nested(), bind, close(), close_all(), commit(), configure(), connection(), delete(), deleted, dirty, execute(), expire(), expire_all(), expunge(), expunge_all(), flush(), get(), get_bind(), get_one(), identity_key(), identity_map, info, invalidate(), is_active, is_modified(), merge(), new, no_autoflush, object_session(), refresh(), remove(), reset(), rollback(), scalar(), scalars(), session_factory, stream(), stream_scalars()
Class signature
class
sqlalchemy.ext.asyncio.async_scoped_session
(typing.Generic
)-
method
sqlalchemy.ext.asyncio.async_scoped_session.
__call__(**kw: Any) _AS ¶ Return the current
AsyncSession
, creating it using thescoped_session.session_factory
if not present.- Parameters:
**kw¶ – Keyword arguments will be passed to the
scoped_session.session_factory
callable, if an existingAsyncSession
is not present. If theAsyncSession
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 ofasync_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 theasync_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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 toSession.add()
will be moved back to the transient state, and will no longer be present within thisSession
.
-
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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 theAsyncSession
class.Proxied for the
AsyncSession
class on behalf of theasync_scoped_session
class.
-
method
sqlalchemy.ext.asyncio.async_scoped_session.
begin() AsyncSessionTransaction ¶ Return an
AsyncSessionTransaction
object.Proxied for the
AsyncSession
class on behalf of theasync_scoped_session
class.The underlying
Session
will perform the “begin” action when theAsyncSessionTransaction
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 theasync_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 theasync_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 theasync_scoped_session
class.See also
Session.close()
- main documentation for “close”Closing - detail on the semantics of
AsyncSession.close()
andAsyncSession.reset()
.
-
async classmethod
sqlalchemy.ext.asyncio.async_scoped_session.
close_all() None ¶ Close all
AsyncSession
sessions.Proxied for the
AsyncSession
class on behalf of theasync_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 toclose_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 theasync_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 thisscoped_session
.
-
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 thisSession
object’s transactional state.Proxied for the
AsyncSession
class on behalf of theasync_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 theasync_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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 theasync_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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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, useSession.expire_all()
.The
Session
object’s default behavior is to expire all state whenever theSession.rollback()
orSession.commit()
methods are called, so that new state can be loaded for the new transaction. For this reason, callingSession.expire()
only makes sense for the specific case that a non-ORM SQL statement was emitted in the current transaction.- Parameters:
See also
Refreshing / Expiring - introductory material
-
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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 theSession.rollback()
orSession.commit()
methods are called, so that new state can be loaded for the new transaction. For this reason, callingSession.expire_all()
is not usually needed, assuming the transaction is isolated.See also
Refreshing / Expiring - introductory material
-
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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
class.This is equivalent to calling
expunge(obj)
on all objects in thisSession
.
-
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 theasync_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 theasync_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 theasync_scoped_session
class.Unlike the
Session.get_bind()
method, this method is currently not used by thisAsyncSession
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 theSession.get_bind()
method. The example below illustrates how to implement customSession.get_bind()
schemes that work withAsyncSession
andAsyncEngine
.The pattern introduced at Custom Vertical Partitioning illustrates how to apply a custom bind-lookup scheme to a
Session
given a set ofEngine
objects. To apply a correspondingSession.get_bind()
implementation for use with aAsyncSession
andAsyncEngine
objects, continue to subclassSession
and apply it toAsyncSession
usingAsyncSession.sync_session_class
. The inner method must continue to returnEngine
instances, which can be acquired from aAsyncEngine
using theAsyncEngine.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 viaAsyncSession.run_sync()
, so routines that wish to run SQL commands inside ofSession.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 theasync_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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 theAsyncSession
class.Proxied for the
AsyncSession
class on behalf of theasync_scoped_session
class.
-
attribute
sqlalchemy.ext.asyncio.async_scoped_session.
info¶ A user-modifiable dictionary.
Proxied for the
AsyncSession
class on behalf of theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
class.The initial value of this dictionary can be populated using the
info
argument to theSession
constructor orsessionmaker
constructor or factory methods. The dictionary here is always local to thisSession
and can be modified independently of all otherSession
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 theasync_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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
class.Changed in version 1.4: The
Session
no longer begins a new transaction immediately, so this attribute will be False when theSession
is first instantiated.“partial rollback” state typically indicates that the flush process of the
Session
has failed, and that theSession.rollback()
method must be emitted in order to fully roll back the transaction.If this
Session
is not in a transaction at all, theSession
will autobegin when it is first used, so in this caseSession.is_active
will return True.Otherwise, if this
Session
is within a transaction, and that transaction has not been rolled back internally, theSession.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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 reportFalse
when tested with this method. This is because the object may have received change events via attribute mutation, thus placing it inSession.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 toTrue
. 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 theactive_history
argument withcolumn_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 theasync_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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 theasync_scoped_session
class.Proxied for the
Session
class on behalf of theAsyncSession
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 theasync_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 theasync_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()
andAsyncSession.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 theasync_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 theasync_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 theasync_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 scalarsAsyncSession.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 theasync_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 theasync_scoped_session
class.- Returns:
an
AsyncScalarResult
object
New in version 1.4.24.
See also
Session.scalars()
- main documentation for scalarsAsyncSession.scalars()
- non streaming version
-
method
- 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 declarativeBase
class where it takes effect for all subclasses. This mixin adds a single new attributeAsyncAttrs.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 theAsyncSession.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
-
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
-
attribute
- class sqlalchemy.ext.asyncio.AsyncSession¶
Asyncio version of
Session
.The
AsyncSession
is a proxy for a traditionalSession
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 customSession
implementations, see theAsyncSession.sync_session_class
parameter.Members
sync_session_class, __init__(), aclose(), add(), add_all(), autoflush, begin(), begin_nested(), close(), close_all(), commit(), connection(), delete(), deleted, dirty, execute(), expire(), expire_all(), expunge(), expunge_all(), flush(), get(), get_bind(), get_nested_transaction(), get_one(), get_transaction(), identity_key(), identity_map, in_nested_transaction(), in_transaction(), info, invalidate(), is_active, is_modified(), merge(), new, no_autoflush, object_session(), refresh(), reset(), rollback(), run_sync(), scalar(), scalars(), stream(), stream_scalars(), sync_session
Class signature
class
sqlalchemy.ext.asyncio.AsyncSession
(sqlalchemy.ext.asyncio.base.ReversibleProxy
)-
attribute
sqlalchemy.ext.asyncio.AsyncSession.
sync_session_class: Type[Session] = <class 'sqlalchemy.orm.session.Session'>¶ The class or callable that provides the underlying
Session
instance for a particularAsyncSession
.At the class level, this attribute is the default value for the
AsyncSession.sync_session_class
parameter. Custom subclasses ofAsyncSession
can override this.At the instance level, this attribute indicates the current class or callable that was used to provide the
Session
instance for thisAsyncSession
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 thesync_session_class
callable directly to instantiate a newSession
. Refer toSession.__init__()
for parameter documentation.- Parameters:
sync_session_class¶ –
A
Session
subclass or other callable which will be used to construct theSession
which will be proxied. This parameter may be used to provide customSession
subclasses. Defaults to theAsyncSession.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 theAsyncSession
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 toSession.add()
will be moved back to the transient state, and will no longer be present within thisSession
.
-
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 theAsyncSession
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 theAsyncSession
class.
-
method
sqlalchemy.ext.asyncio.AsyncSession.
begin() AsyncSessionTransaction ¶ Return an
AsyncSessionTransaction
object.The underlying
Session
will perform the “begin” action when theAsyncSessionTransaction
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()
andAsyncSession.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 toclose_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 thisSession
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 theAsyncSession
class.
-
attribute
sqlalchemy.ext.asyncio.AsyncSession.
dirty¶ The set of all persistent instances considered dirty.
Proxied for the
Session
class on behalf of theAsyncSession
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 theAsyncSession
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, useSession.expire_all()
.The
Session
object’s default behavior is to expire all state whenever theSession.rollback()
orSession.commit()
methods are called, so that new state can be loaded for the new transaction. For this reason, callingSession.expire()
only makes sense for the specific case that a non-ORM SQL statement was emitted in the current transaction.- Parameters:
See also
Refreshing / Expiring - introductory material
-
method
sqlalchemy.ext.asyncio.AsyncSession.
expire_all() None ¶ Expires all persistent instances within this Session.
Proxied for the
Session
class on behalf of theAsyncSession
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 theSession.rollback()
orSession.commit()
methods are called, so that new state can be loaded for the new transaction. For this reason, callingSession.expire_all()
is not usually needed, assuming the transaction is isolated.See also
Refreshing / Expiring - introductory material
-
method
sqlalchemy.ext.asyncio.AsyncSession.
expunge(instance: object) None ¶ Remove the instance from this
Session
.Proxied for the
Session
class on behalf of theAsyncSession
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 theAsyncSession
class.This is equivalent to calling
expunge(obj)
on all objects in thisSession
.
-
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 thisAsyncSession
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 theSession.get_bind()
method. The example below illustrates how to implement customSession.get_bind()
schemes that work withAsyncSession
andAsyncEngine
.The pattern introduced at Custom Vertical Partitioning illustrates how to apply a custom bind-lookup scheme to a
Session
given a set ofEngine
objects. To apply a correspondingSession.get_bind()
implementation for use with aAsyncSession
andAsyncEngine
objects, continue to subclassSession
and apply it toAsyncSession
usingAsyncSession.sync_session_class
. The inner method must continue to returnEngine
instances, which can be acquired from aAsyncEngine
using theAsyncEngine.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 viaAsyncSession.run_sync()
, so routines that wish to run SQL commands inside ofSession.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, orNone
.
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, orNone
.
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 theAsyncSession
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 theAsyncSession
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 theAsyncSession
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 theAsyncSession
class.New in version 1.4.
See also
-
attribute
sqlalchemy.ext.asyncio.AsyncSession.
info¶ A user-modifiable dictionary.
Proxied for the
Session
class on behalf of theAsyncSession
class.The initial value of this dictionary can be populated using the
info
argument to theSession
constructor orsessionmaker
constructor or factory methods. The dictionary here is always local to thisSession
and can be modified independently of all otherSession
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 theAsyncSession
class.Changed in version 1.4: The
Session
no longer begins a new transaction immediately, so this attribute will be False when theSession
is first instantiated.“partial rollback” state typically indicates that the flush process of the
Session
has failed, and that theSession.rollback()
method must be emitted in order to fully roll back the transaction.If this
Session
is not in a transaction at all, theSession
will autobegin when it is first used, so in this caseSession.is_active
will return True.Otherwise, if this
Session
is within a transaction, and that transaction has not been rolled back internally, theSession.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 theAsyncSession
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 reportFalse
when tested with this method. This is because the object may have received change events via attribute mutation, thus placing it inSession.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 toTrue
. 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 theactive_history
argument withcolumn_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 theAsyncSession
class.
-
attribute
sqlalchemy.ext.asyncio.AsyncSession.
no_autoflush¶ Return a context manager that disables autoflush.
Proxied for the
Session
class on behalf of theAsyncSession
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 theAsyncSession
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()
andAsyncSession.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
-
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 scalarsAsyncSession.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 scalarsAsyncSession.scalars()
- non streaming version
-
attribute
sqlalchemy.ext.asyncio.AsyncSession.
sync_session: Session¶ Reference to the underlying
Session
thisAsyncSession
proxies requests towards.This instance can be used as an event target.
-
attribute
- 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()
andAsyncSessionTransaction.rollback()
, as well as use as an async context manager.New in version 1.4.
Members
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
.
-
method