Asynchronous I/O (asyncio) — SQLAlchemy 2.0 Documentation (original) (raw)

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

New in version 1.4.

Warning

Please read Asyncio Platform Installation Notes (Including Apple M1) for important platform installation notes for many platforms, including Apple M1 Architecture.

Asyncio Platform Installation Notes (Including Apple M1)¶

The asyncio extension requires Python 3 only. It also depends upon the greenlet library. This dependency is installed by default on common machine platforms including:

x86_64 aarch64 ppc64le amd64 win32

For the above platforms, greenlet is known to supply pre-built wheel files. For other platforms, greenlet does not install by default; the current file listing for greenlet can be seen atGreenlet - Download Files. Note that there are many architectures omitted, including Apple M1.

To install SQLAlchemy while ensuring the greenlet dependency is present regardless of what platform is in use, the[asyncio] setuptools extramay be installed as follows, which will include also instruct pip to install greenlet:

pip install sqlalchemy[asyncio]

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

Synopsis - Core¶

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

import asyncio

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

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

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

asyncio.run(async_main())

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

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

Tip

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

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

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

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

Synopsis - ORM¶

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

Within the default mode of use, special care must be taken to avoid lazy loading or other expired-attribute access involving ORM relationships and column attributes; the next section Preventing Implicit IO when Using AsyncSession details this.

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

from future import annotations

import asyncio import datetime from typing import List

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

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

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

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

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

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

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

asyncio.run(async_main())

BEGIN (implicit) ... CREATE TABLE a ( id INTEGER NOT NULL, data VARCHAR NOT NULL, create_date DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL, PRIMARY KEY (id) ) ... CREATE TABLE b ( id INTEGER NOT NULL, a_id INTEGER NOT NULL, data VARCHAR NOT NULL, PRIMARY KEY (id), FOREIGN KEY(a_id) REFERENCES a (id) ) ... COMMIT BEGIN (implicit) INSERT INTO a (data) VALUES (?) RETURNING id, create_date [...] ('a1',) ... INSERT INTO b (a_id, data) VALUES (?, ?) RETURNING id [...] (1, 'b2') ... COMMIT BEGIN (implicit) SELECT a.id, a.data, a.create_date FROM a ORDER BY a.id [...] () SELECT b.a_id AS b_a_id, b.id AS b_id, b.data AS b_data FROM b WHERE b.a_id IN (?, ?, ?) [...] (1, 2, 3) <A object at ...> a1 created at: ... b1 b2 <A object at ...> a2 created at: ... <A object at ...> a3 created at: ... b3 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 b1 b2

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

Using AsyncSession with Concurrent Tasks¶

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

See the section Is the Session thread-safe? Is AsyncSession safe to share in concurrent tasks? for a general description of the Session and AsyncSession with regards to how they should be used with concurrent workloads.

Preventing Implicit IO when Using AsyncSession¶

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

Attributes that are lazy-loading relationships, deferred columns or expressions, or are being accessed in expiration scenarios can take advantage of the AsyncAttrs mixin. This mixin, when added to a specific class or more generally to the Declarative Base superclass, provides an accessor AsyncAttrs.awaitable_attrswhich delivers any attribute as an awaitable:
from future import annotations
from typing import List
from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import relationship
class Base(AsyncAttrs, DeclarativeBase):
pass
class A(Base):
tablename = "a"
... rest of mapping ...
bs: Mapped[List[B]] = relationship()
class B(Base):
tablename = "b"
... rest of mapping ...
Accessing the A.bs collection on newly loaded instances of A when eager loading is not in use will normally use lazy loading, which in order to succeed will usually emit IO to the database, which will fail under asyncio as no implicit IO is allowed. To access this attribute directly under asyncio without any prior loading operations, the attribute can be accessed as an awaitable by indicating the AsyncAttrs.awaitable_attrsprefix:
a1 = (await session.scalars(select(A))).one()
for b1 in await a1.awaitable_attrs.bs:
print(b1)
The AsyncAttrs mixin provides a succinct facade over the internal approach that’s also used by theAsyncSession.run_sync() method.
New in version 2.0.13.
Collections can be replaced with write only collections that will never emit IO implicitly, by using the Write Only Relationships feature in SQLAlchemy 2.0. Using this feature, collections are never read from, only queried using explicit SQL calls. See the exampleasync_orm_writeonly.py in the Asyncio Integration section for an example of write-only collections used with asyncio.
When using write only collections, the program’s behavior is simple and easy to predict regarding collections. However, the downside is that there is not any built-in system for loading many of these collections all at once, which instead would need to be performed manually. Therefore, many of the bullets below address specific techniques when using traditional lazy-loaded relationships with asyncio, which requires more care.
If not using AsyncAttrs, relationships can be declared with lazy="raise" so that by default they will not attempt to emit SQL. In order to load collections, eager loading would be used instead.
The most useful eager loading strategy is theselectinload() eager loader, which is employed in the previous example in order to eagerly load the A.bs collection within the scope of theawait session.execute() call:
stmt = select(A).options(selectinload(A.bs))
When constructing new objects, collections are always assigned a default, empty collection, such as a list in the above example:
This allows the .bs collection on the above A object to be present and readable when the A object is flushed; otherwise, when the A is flushed, .bs would be unloaded and would raise an error on access.
The AsyncSession is configured usingSession.expire_on_commit set to False, so that we may access attributes on an object subsequent to a call toAsyncSession.commit(), as in the line at the end where we access an attribute:

create AsyncSession with expire_on_commit=False

async_session = AsyncSession(engine, expire_on_commit=False)

sessionmaker version

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

Other guidelines include:

Methods like AsyncSession.expire() should be avoided in favor ofAsyncSession.refresh(); if expiration is absolutely needed. Expiration should generally not be needed asSession.expire_on_commitshould normally be set to False when using asyncio.
A lazy-loaded relationship can be loaded explicitly under asyncio usingAsyncSession.refresh(), if the desired attribute name is passed explicitly toSession.refresh.attribute_names, e.g.:

assume a_obj is an A that has lazy loaded A.bs collection

a_obj = await async_session.get(A, [1])

force the collection to load by naming it in attribute_names

await async_session.refresh(a_obj, ["bs"])

collection is present

print(f"bs collection: {a_obj.bs}")
It’s of course preferable to use eager loading up front in order to have collections already set up without the need to lazy-load.
New in version 2.0.4: Added support forAsyncSession.refresh() and the underlyingSession.refresh() method to force lazy-loaded relationships to load, if they are named explicitly in theSession.refresh.attribute_names parameter. In previous versions, the relationship would be silently skipped even if named in the parameter.

Avoid using the all cascade option documented at Cascadesin favor of listing out the desired cascade features explicitly. Theall cascade option implies among others the refresh-expiresetting, which means that the AsyncSession.refresh() method will expire the attributes on related objects, but not necessarily refresh those related objects assuming eager loading is not configured within therelationship(), leaving them in an expired state.
Appropriate loader options should be employed for deferred()columns, if used at all, in addition to that of relationship()constructs as noted above. See Limiting which Columns Load with Column Deferral for background on deferred column loading.
The “dynamic” relationship loader strategy described atDynamic Relationship Loaders is not compatible by default with the asyncio approach. It can be used directly only if invoked within theAsyncSession.run_sync() method described atRunning Synchronous Methods and Functions under asyncio, or by using its .statement attribute to obtain a normal select:
user = await session.get(User, 42)
addresses = (await session.scalars(user.addresses.statement)).all()
stmt = user.addresses.statement.where(Address.email_address.startswith("patrick"))
addresses_filter = (await session.scalars(stmt)).all()
The write only technique, introduced in version 2.0 of SQLAlchemy, is fully compatible with asyncio and should be preferred.
If using asyncio with a database that does not support RETURNING, such as MySQL 8, server default values such as generated timestamps will not be available on newly flushed objects unless theMapper.eager_defaults option is used. In SQLAlchemy 2.0, this behavior is applied automatically to backends like PostgreSQL, SQLite and MariaDB which use RETURNING to fetch new values when rows are INSERTed.

Running Synchronous Methods and Functions under asyncio¶

Deep Alchemy

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

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

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

import asyncio

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

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

"""

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

stmt = select(A)

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

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

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

a1.data = "new data"

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

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

    await session.run_sync(fetch_and_update_objects)

    await session.commit()

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

asyncio.run(async_main())

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

unlike when using gevent, we can continue to use the standard Python asyncio event loop, or any custom event loop, without the need to integrate into the gevent event loop.
There is no “monkeypatching” whatsoever. The above example makes use of a real asyncio driver and the underlying SQLAlchemy connection pool is also using the Python built-in asyncio.Queue for pooling connections.
The program can freely switch between async/await code and contained functions that use sync code with virtually no performance penalty. There is no “thread executor” or any additional waiters or synchronization in use.
The underlying network drivers are also using pure Python asyncio concepts, no third party networking libraries as gevent and eventletprovides are in use.

Using events with the asyncio extension¶

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

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

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

Events can be registered at the instance level (e.g. a specificAsyncEngine instance) by associating the event with thesync attribute that refers to the proxied object. For example to register the PoolEvents.connect() event against anAsyncEngine instance, use itsAsyncEngine.sync_engine attribute as target. Targets include:
To register an event at the class level, targeting all instances of the same type (e.g. all AsyncSession instances), use the corresponding sync-style class. For example to register theSessionEvents.before_commit() event against theAsyncSession class, use the Session class as the target.
To register at the sessionmaker level, combine an explicitsessionmaker with an async_sessionmakerusing async_sessionmaker.sync_session_class, and associate events with the sessionmaker.

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

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

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

Core Events on AsyncEngine
In this example, we access the AsyncEngine.sync_engineattribute of AsyncEngine as the target forConnectionEvents and PoolEvents:
import asyncio
from sqlalchemy import event
from sqlalchemy import text
from sqlalchemy.engine import Engine
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost:5432/test")

connect event on instance of Engine

@event.listens_for(engine.sync_engine, "connect")
def my_on_connect(dbapi_con, connection_record):
print("New DBAPI connection:", dbapi_con)
cursor = dbapi_con.cursor()
# sync style API use for adapted DBAPI connection / cursor
cursor.execute("select 'execute from event'")
print(cursor.fetchone()[0])

before_execute event on all Engine instances

@event.listens_for(Engine, "before_execute")
def my_before_execute(
conn,
clauseelement,
multiparams,
params,
execution_options,
):
print("before execute!")
async def go():
async with engine.connect() as conn:
await conn.execute(text("select 1"))
await engine.dispose()
asyncio.run(go())
Output:
New DBAPI connection: <AdaptedConnection <asyncpg.connection.Connection object at 0x7f33f9b16960>>
execute from event
before execute!

ORM Events on AsyncSession
In this example, we access AsyncSession.sync_session as the target for SessionEvents:
import asyncio
from sqlalchemy import event
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import Session
engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost:5432/test")
session = AsyncSession(engine)

before_commit event on instance of Session

@event.listens_for(session.sync_session, "before_commit")
def my_before_commit(session):
print("before commit!")
# sync style API use on Session
connection = session.connection()
# sync style API use on Connection
result = connection.execute(text("select 'execute from event'"))
print(result.first())

after_commit event on all Session instances

@event.listens_for(Session, "after_commit")
def my_after_commit(session):
print("after commit!")
async def go():
await session.execute(text("select 1"))
await session.commit()
await session.close()
await engine.dispose()
asyncio.run(go())
Output:
before commit!
execute from event
after commit!

ORM Events on async_sessionmaker
For this use case, we make a sessionmaker as the event target, then assign it to the async_sessionmaker using the async_sessionmaker.sync_session_class parameter:
import asyncio
from sqlalchemy import event
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.orm import sessionmaker
sync_maker = sessionmaker()
maker = async_sessionmaker(sync_session_class=sync_maker)
@event.listens_for(sync_maker, "before_commit")
def before_commit(session):
print("before commit")
async def main():
async_session = maker()
await async_session.commit()
asyncio.run(main())
Output:

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

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

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

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

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

engine = create_async_engine(...)

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

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

New in version 1.4.30.

Using multiple asyncio event loops¶

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

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

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

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

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

Using asyncio scoped session¶

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

Tip

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

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

from asyncio import current_task

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

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

Warning

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

Warning

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

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

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

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

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

New in version 1.4.19.

Using the Inspector to inspect schema objects¶

SQLAlchemy does not yet offer an asyncio version of theInspector (introduced at Fine Grained Reflection with Inspector), however the existing interface may be used in an asyncio context by leveraging the AsyncConnection.run_sync() method ofAsyncConnection:

import asyncio

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

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

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

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

asyncio.run(async_main())

Engine API Documentation¶

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

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

Create a new async engine instance.

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

New in version 1.4.

Parameters:

async_creator¶ –

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

This parameter is the asyncio equivalent of thecreate_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 of engine_from_config().

New in version 1.4.29.

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

Create a new async engine instance.

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

New in version 2.0.10.

class sqlalchemy.ext.asyncio.AsyncEngine¶

An asyncio proxy for a Engine.

AsyncEngine is acquired using 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 anAsyncConnection 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 the AsyncEngine class.

This applies only to the built-in cache that is established via the create_engine.query_cache_size parameter. It will not impact any dictionary caches that were passed via 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 its AsyncConnection.start()method.

attribute sqlalchemy.ext.asyncio.AsyncEngine.dialect¶

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

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

Dispose of the connection pool used by thisAsyncEngine.

Parameters:

close¶ –

if left at its default of True, has the effect of fully closing all currently checked indatabase connections. Connections that are still checked out will not be closed, however they will no longer be associated with this Engine, so when they are closed individually, eventually 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.

attribute sqlalchemy.ext.asyncio.AsyncEngine.driver¶

Driver name of the Dialectin use by this Engine.

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

attribute sqlalchemy.ext.asyncio.AsyncEngine.echo¶

When True, enable log output for this element.

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

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

attribute sqlalchemy.ext.asyncio.AsyncEngine.engine¶

Returns this Engine.

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

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

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

Return a new AsyncEngine that will 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 the AsyncEngine class.

attribute sqlalchemy.ext.asyncio.AsyncEngine.name¶

String name of the Dialectin use by this Engine.

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

attribute sqlalchemy.ext.asyncio.AsyncEngine.pool¶

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

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

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

attribute sqlalchemy.ext.asyncio.AsyncEngine.sync_engine_: Engine_¶

Reference to the sync-style Engine 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 the AsyncEngine class.

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

attribute sqlalchemy.ext.asyncio.AsyncEngine.url¶

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

class sqlalchemy.ext.asyncio.AsyncConnection¶

An asyncio proxy for a Connection.

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

from sqlalchemy.ext.asyncio import create_async_engine

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

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

New in version 1.4.

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.aclosingcontext manager function.

New in version 2.0.20.

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

Begin a transaction prior to autobegin occurring.

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

Begin a nested transaction and return a transaction handle.

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

Close this AsyncConnection.

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

attribute sqlalchemy.ext.asyncio.AsyncConnection.closed¶

Return True if this connection is closed.

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

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

Commit the transaction that is currently in progress.

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

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

attribute sqlalchemy.ext.asyncio.AsyncConnection.connection¶

Not implemented for async; callAsyncConnection.get_raw_connection().

attribute sqlalchemy.ext.asyncio.AsyncConnection.default_isolation_level¶

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

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

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

Calling this accessor does not invoke any new SQL queries.

attribute sqlalchemy.ext.asyncio.AsyncConnection.dialect¶

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

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

Executes a driver-level SQL string and return bufferedResult.

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

Parameters:

object¶ –
The statement to be executed. This is always an object that is in both the ClauseElement andExecutable hierarchies, including:
- Select
- Insert, Update,Delete
- TextClause andTextualSelect
- 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 DBAPI cursor.execute()method will be used.
execution_options¶ – optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by Connection.execution_options().

Returns:

a Result object.

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

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

This returns this AsyncConnection object with the new options added.

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

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

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

This makes use of the underlying synchronous connection’sConnection.get_nested_transaction() method to get the current Transaction, 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 thisAsyncConnection.

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

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

Return an AsyncTransaction representing the current transaction, if any.

This makes use of the underlying synchronous connection’sConnection.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 underlying Connection.

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

This attribute is only available if the AsyncConnection is currently connected. If the AsyncConnection.closed attribute is True, then accessing this attribute will 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 the AsyncConnection class.

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

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

Roll back the transaction that is currently in progress.

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

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

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

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

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

E.g.:

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

:param conn: a Core SQLAlchemy Connection, used synchronously

:return: an optional return value is supported

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

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

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

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

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

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 theResult.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 theResult.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 Python with: block.

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

Execute a statement and return an awaitable yielding aAsyncResult 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 an async with fn(): orawait fn() style.

New in version 2.0.0b3: added context manager support

Returns:

an awaitable object that will yield anAsyncResult object.

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

Execute a statement and return an awaitable yielding aAsyncScalarResult object.

E.g.:

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

This method is shorthand for invoking theAsyncResult.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 an async with fn(): orawait fn() style.

New in version 2.0.0b3: added context manager support

Returns:

an awaitable object that will yield anAsyncScalarResult object.

New in version 1.4.24.

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.

class sqlalchemy.ext.asyncio.AsyncTransaction¶

An asyncio proxy for a Transaction.

Class signature

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

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

Close this AsyncTransaction.

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

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

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

Commit this AsyncTransaction.

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

Roll back this AsyncTransaction.

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

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

Result Set API Documentation¶

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

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

class sqlalchemy.ext.asyncio.AsyncResult¶

An asyncio wrapper around a Result object.

The AsyncResult only applies to statement executions that use a server-side cursor. It is returned only from theAsyncConnection.stream() andAsyncSession.stream() methods.

Note

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

New in version 1.4.

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

Close this result.

attribute sqlalchemy.ext.asyncio.AsyncResult.closed¶

inherited from the AsyncCommon.closed attribute of AsyncCommon

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

New in version 2.0.0b3.

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

Establish the columns that should be returned in each row.

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

method sqlalchemy.ext.asyncio.AsyncResult.async fetchall() → Sequence[Row[_TP]]¶

A synonym for the AsyncResult.all() method.

New in version 2.0.

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

Fetch many rows.

When all rows are exhausted, returns an empty list.

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

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

Returns:

a list of Row objects.

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

Fetch one row.

When all rows are exhausted, returns None.

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

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

Returns:

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

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

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

Closes the result set and discards remaining rows.

Note

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

Additionally, in contrast to the behavior of the legacy ORMQuery.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.

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 thisAsyncResult when invoked.

The callable object returned is an instance ofFrozenResult.

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 FrozenResultis retrieved from a cache, it can be called any number of times where it will produce a new Result object each time against its stored set of rows.

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

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

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

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

The view also can be tested for key containment using the Pythonin 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 ofAsyncMappingResult.

When this filter is applied, fetching rows will returnRowMapping objects instead of Rowobjects.

Returns:

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

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

Return exactly one row or raise an exception.

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

Note

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

New in version 1.4.

Returns:

The first Row.

Raises:

MultipleResultsFound, NoResultFound

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

Return at most one result or raise an exception.

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

New in version 1.4.

Returns:

The first Row or None if no row is available.

Raises:

MultipleResultsFound

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

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

An async iterator is returned:

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

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

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

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

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

Returns None if there are no rows to fetch.

No validation is performed to test if additional rows remain.

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

Returns:

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

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

Return exactly one scalar result or raise an exception.

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

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

Return exactly one scalar result or None.

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

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

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

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

Parameters:

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

Returns:

a new AsyncScalarResult filtering object referring to this AsyncResult object.

attribute sqlalchemy.ext.asyncio.AsyncResult.t¶

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

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

New in version 2.0.

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

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

This method returns the same AsyncResult object at runtime, however annotates as returning a AsyncTupleResultobject that will indicate to PEP 484 typing tools that plain typedTuple instances are returned rather than rows. This allows tuple unpacking and __getitem__ access of Rowobjects 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.

ORM Session API Documentation¶

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

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

Return the AsyncSession to which the given instance belongs.

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

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

This functionality is also available from theInstanceState.async_session accessor.

Parameters:

instance¶ – an ORM mapped instance

Returns:

an AsyncSession object, or None.

New in version 1.4.18.

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

Return the AsyncSession which is proxying the givenSession object, if any.

Parameters:

session¶ – a Session instance.

Returns:

a AsyncSession instance, or None.

New in version 1.4.18.

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

Close all AsyncSession sessions.

New in version 2.0.23.

using example from "Custom Vertical Partitioning"

import random

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

construct async engines w/ async drivers

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

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

apply to AsyncSession using sync_session_class

AsyncSessionMaker = async_sessionmaker(sync_session_class=RoutingSession)

The Session.get_bind() method is called in a non-asyncio, implicitly non-blocking context in the same manner as ORM event hooks and functions that are invoked via AsyncSession.run_sync(), so routines that wish to run SQL commands inside 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 the async_scoped_session class.

Raises NoResultFound if the query selects no rows.

..versionadded: 2.0.22

Return an identity key.

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

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

This is an alias of identity_key().

attribute sqlalchemy.ext.asyncio.async_scoped_session.identity_map¶

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

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

attribute sqlalchemy.ext.asyncio.async_scoped_session.info¶

A user-modifiable dictionary.

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

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

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

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

Close this Session, using connection invalidation.

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

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

attribute sqlalchemy.ext.asyncio.async_scoped_session.is_active¶

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

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

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

Changed in version 1.4: The Session no longer begins a new transaction immediately, so this attribute will be False when the Session is first instantiated.

“partial rollback” state typically indicates that the flush process of the Session has failed, and that 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 case Session.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 the async_scoped_session class.

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

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

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

E.g.:

return session.is_modified(someobject)

A few caveats to this method apply:

Instances present in the Session.dirty collection may report False when tested with this method. This is because the object may have received change events via attribute mutation, thus placing it in Session.dirty, but ultimately the state is the same as that loaded from the database, resulting in no net change here.
Scalar attributes may not have recorded the previously set value when a new value was applied, if the attribute was not loaded, or was expired, at the time the new value was received - in these cases, the attribute is assumed to have a change, even if there is ultimately no net change against its database value. SQLAlchemy in most cases does not need the “old” value when a set event occurs, so it skips the expense of a SQL call if the old value isn’t present, based on the assumption that an UPDATE of the scalar value is usually needed, and in those few cases where it isn’t, is less expensive on average than issuing a defensive SELECT.
The “old” value is fetched unconditionally upon set only if the attribute container has the active_history flag set to True. This flag is set typically for primary key attributes and scalar object references that are not a simple many-to-one. To set this flag for any arbitrary mapped column, use the active_historyargument with column_property().

Parameters:

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

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

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

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

attribute sqlalchemy.ext.asyncio.async_scoped_session.new¶

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

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

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

attribute sqlalchemy.ext.asyncio.async_scoped_session.no_autoflush¶

Return a context manager that disables autoflush.

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

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

e.g.:

with session.no_autoflush:

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

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

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

Return the Session to which an object belongs.

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

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

This is an alias of object_session().

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

Expire and refresh the attributes on the given instance.

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

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

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

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 thisSession, resetting the session to its initial state.

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

New in version 2.0.22.

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

Rollback the current transaction in progress.

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

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

Execute a statement and return a scalar result.

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

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

Execute a statement and return scalar results.

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

Returns:

a ScalarResult object

New in version 1.4.24: Added AsyncSession.scalars()

New in version 1.4.26: Addedasync_scoped_session.scalars()

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

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

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

Execute a statement and return a stream of scalar results.

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

Returns:

an AsyncScalarResult object

New in version 1.4.24.

class sqlalchemy.ext.asyncio.AsyncAttrs¶

Mixin class which provides an awaitable accessor for all attributes.

E.g.:

from future import annotations

from typing import List

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

class Base(AsyncAttrs, DeclarativeBase): pass

class A(Base): tablename = "a"

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

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

In the above example, the AsyncAttrs mixin is applied to the declarative Base class where it takes effect for all subclasses. This mixin adds a single new 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.

attribute sqlalchemy.ext.asyncio.AsyncAttrs.awaitable_attrs¶

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

e.g.:

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

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

class sqlalchemy.ext.asyncio.AsyncSession¶

Asyncio version of Session.

The AsyncSession is a proxy for a 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 custom Sessionimplementations, 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 theAsyncSession.sync_session_class parameter. Custom subclasses of AsyncSession can override this.

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

New in version 1.4.24.

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

Construct a new AsyncSession.

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

Parameters:

sync_session_class¶ –

A Session subclass or other callable which will be used to construct the Session which will be proxied. This parameter may be used to provide custom Sessionsubclasses. 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.aclosingcontext manager function.

New in version 2.0.20.

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

Place an object into this Session.

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

Objects that are in the transient state when passed to theSession.add() method will move to thepending 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 theSession.add() method will move to the persistentstate directly.

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

Add the given collection of instances to this Session.

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

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

attribute sqlalchemy.ext.asyncio.AsyncSession.autoflush¶

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

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

Return an AsyncSessionTransaction object.

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

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

For a general description of ORM begin, seeSession.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, seeSession.begin_nested().

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

Close out the transactional resources and ORM objects used by thisAsyncSession.

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

Close all AsyncSession sessions.

Deprecated since version 2.0: The AsyncSession.close_all() method is deprecated and will be removed in a future release. Please refer to close_all_sessions().

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

Commit the current transaction in progress.

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

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

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

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

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.

attribute sqlalchemy.ext.asyncio.AsyncSession.deleted¶

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

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

attribute sqlalchemy.ext.asyncio.AsyncSession.dirty¶

The set of all persistent instances considered dirty.

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

E.g.:

some_mapped_object in session.dirty

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

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

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

Expire the attributes on an instance.

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

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

Parameters:

instance¶ – The instance to be refreshed.
attribute_names¶ – optional list of string attribute names indicating a subset of attributes to be expired.

method sqlalchemy.ext.asyncio.AsyncSession.expire_all() → None¶

Expires all persistent instances within this Session.

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

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

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

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

Remove the instance from this Session.

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

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

method sqlalchemy.ext.asyncio.AsyncSession.expunge_all() → None¶

Remove all object instances from this Session.

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

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

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

Flush all the object changes to the database.

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.

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

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

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

apply to AsyncSession using sync_session_class

AsyncSessionMaker = async_sessionmaker(sync_session_class=RoutingSession)

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 NoResultFound if the query selects no rows.

..versionadded: 2.0.22

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.

Return an identity key.

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

This is an alias of identity_key().

attribute sqlalchemy.ext.asyncio.AsyncSession.identity_map¶

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

method sqlalchemy.ext.asyncio.AsyncSession.in_nested_transaction() → bool¶

Return True if this Session has begun a nested transaction, e.g. SAVEPOINT.

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

New in version 1.4.

method sqlalchemy.ext.asyncio.AsyncSession.in_transaction() → bool¶

Return True if this Session has begun a transaction.

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

New in version 1.4.

attribute sqlalchemy.ext.asyncio.AsyncSession.info¶

A user-modifiable dictionary.

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

method sqlalchemy.ext.asyncio.AsyncSession.async invalidate() → None¶

Close this Session, using connection invalidation.

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

attribute sqlalchemy.ext.asyncio.AsyncSession.is_active¶

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

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

Changed in version 1.4: The Session no longer begins a new transaction immediately, so this attribute will be False when the Session is first instantiated.

“partial rollback” state typically indicates that the flush process of the Session has failed, and that 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 case Session.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 the AsyncSession class.

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

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

E.g.:

return session.is_modified(someobject)

A few caveats to this method apply:

Instances present in the Session.dirty collection may report False when tested with this method. This is because the object may have received change events via attribute mutation, thus placing it in Session.dirty, but ultimately the state is the same as that loaded from the database, resulting in no net change here.
Scalar attributes may not have recorded the previously set value when a new value was applied, if the attribute was not loaded, or was expired, at the time the new value was received - in these cases, the attribute is assumed to have a change, even if there is ultimately no net change against its database value. SQLAlchemy in most cases does not need the “old” value when a set event occurs, so it skips the expense of a SQL call if the old value isn’t present, based on the assumption that an UPDATE of the scalar value is usually needed, and in those few cases where it isn’t, is less expensive on average than issuing a defensive SELECT.
The “old” value is fetched unconditionally upon set only if the attribute container has the active_history flag set to True. This flag is set typically for primary key attributes and scalar object references that are not a simple many-to-one. To set this flag for any arbitrary mapped column, use the active_historyargument with column_property().

Parameters:

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

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

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

attribute sqlalchemy.ext.asyncio.AsyncSession.new¶

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

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

attribute sqlalchemy.ext.asyncio.AsyncSession.no_autoflush¶

Return a context manager that disables autoflush.

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

e.g.:

with session.no_autoflush:

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

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

Return the Session to which an object belongs.

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

This is an alias of object_session().

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

Expire and refresh the attributes on the given instance.

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

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

method sqlalchemy.ext.asyncio.AsyncSession.async reset() → None¶

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

New in version 2.0.22.

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

Rollback the current transaction in progress.

method sqlalchemy.ext.asyncio.AsyncSession.async run_sync(fn: ~typing.Callable[[~typing_extensions.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

Asynchronous I/O (asyncio) — SQLAlchemy 2.0 Documentation (original) (raw)

Asyncio Platform Installation Notes (Including Apple M1)¶

Synopsis - Core¶

Synopsis - ORM¶

Using AsyncSession with Concurrent Tasks¶

Preventing Implicit IO when Using AsyncSession¶

... rest of mapping ...

... rest of mapping ...

create AsyncSession with expire_on_commit=False

sessionmaker version

assume a_obj is an A that has lazy loaded A.bs collection

force the collection to load by naming it in attribute_names

collection is present

Running Synchronous Methods and Functions under asyncio¶

Using events with the asyncio extension¶

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

connect event on instance of Engine

before_execute event on all Engine instances

before_commit event on instance of Session

after_commit event on all Session instances

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

Using multiple asyncio event loops¶

Using asyncio scoped session¶

Using the Inspector to inspect schema objects¶

Engine API Documentation¶

run metadata.create_all(conn) with a sync-style Connection,

proxied into an awaitable

Result Set API Documentation¶

ORM Session API Documentation¶

using example from "Custom Vertical Partitioning"

construct async engines w/ async drivers

apply to AsyncSession using sync_session_class

use the lazy loader on a1.bs via the .awaitable_attrs

interface, so that it may be awaited

using example from "Custom Vertical Partitioning"

construct async engines w/ async drivers

apply to AsyncSession using sync_session_class

use the lazy loader on `a1.bs` via the `.awaitable_attrs`