using example from “Custom Vertical Partitioning”

Asynchronous I/O (asyncio)

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.

Synopsis - Core

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

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(
        "postgresql+asyncpg://scott:tiger@localhost/test",
        echo=True,
    )
    async with engine.begin() as conn:
        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())

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

Tip

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

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

async with engine.connect() as conn:
    async_result = await conn.stream(select(t1))
    async for row in async_result:
        print("row: %s" % (row,))

Synopsis - ORM

Using 2.0 style querying, the 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 details this. The example below illustrates a complete example including mapper and session configuration:

from __future__ import annotations
import asyncio
import datetime
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import select
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(DeclarativeBase):
    pass
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()
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]
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(), B()], data="a1"),
                    A(bs=[B()], data="a2"),
                    A(bs=[B(), B()], data="a3"),
                ]
            )
async def select_and_update_objects(
    async_session: async_sessionmaker[AsyncSession],
) -> None:
    async with async_session() as session:
        stmt = select(A).options(selectinload(A.bs))
        result = await session.execute(stmt)
        for a1 in result.scalars():
            print(a1)
            print(f"created at: {a1.create_date}")
            for b1 in a1.bs:
                print(b1)
        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)
async def async_main() -> None:
    engine = create_async_engine(
        "postgresql+asyncpg://scott:tiger@localhost/test",
        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())

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

Using traditional asyncio, the application needs to avoid any points at which IO-on-attribute access may occur. Above, the following measures are taken to prevent this:

The selectinload() eager loader is employed in order to eagerly load the A.bs collection within the scope of the await session.execute() call:
```
stmt = select(A).options(selectinload(A.bs))
```
If the default loader strategy of “lazyload” were left in place, the access of the A.bs attribute would raise an asyncio exception. There are a variety of ORM loader options available, which may be configured at the default mapping level or used on a per-query basis, documented at .

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

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

The value on the created_at column will not be refreshed by default after an INSERT; instead, it is normally expired so that it can be loaded when needed. Similar behavior applies to a column where the parameter is assigned to a SQL expression object. To access this value with asyncio, it has to be refreshed within the flush process, which is achieved by setting the Mapper.eager_defaults parameter on the mapping:
```
class A(Base):
    # ...
    # column with a server_default, or SQL expression default
    create_date = mapped_column(DateTime, server_default=func.now())
    # add this so that it can be accessed
    __mapper_args__ = {"eager_defaults": True}
```

Other guidelines include:

Methods like should be avoided in favor of AsyncSession.refresh()
Avoid using the all cascade option documented at in favor of listing out the desired cascade features explicitly. The all cascade option implies among others the refresh-expire setting, which means that the method will expire the attributes on related objects, but not necessarily refresh those related objects assuming eager loading is not configured within the relationship(), leaving them in an expired state.
Appropriate loader options should be employed for columns, if used at all, in addition to that of relationship() constructs as noted above. See for background on deferred column loading.
The “dynamic” relationship loader strategy described at Dynamic Relationship Loaders is not compatible by default with the asyncio approach. It can be used directly only if invoked within the method described at Running 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 technique, introduced in version 2.0 of SQLAlchemy, is fully compatible with asyncio and should be preferred.

See also

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

Deep Alchemy

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

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

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
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 eventlet provides are in use.

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

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

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

Events can be registered at the instance level (e.g. a specific instance) by associating the event with the sync attribute that refers to the proxied object. For example to register the PoolEvents.connect() event against an instance, use its AsyncEngine.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 instances), use the corresponding sync-style class. For example to register the SessionEvents.before_commit() event against the class, use the Session class as the target.
To register at the level, combine an explicit sessionmaker with an using async_sessionmaker.sync_session_class, and associate events with the .

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 , the object is a pep-249 compliant “connection” object which will adapt sync-style calls into the asyncio driver.

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

Core Events on AsyncEngine

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

``` 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,
    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 :

``` 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 using the async_sessionmaker.sync_session_class parameter:

``` import asyncio

from sqlalchemy import event from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy.orm import sessionmaker

sync_maker = sessionmaker() maker = async_sessionmaker(sync_session_class=sync_maker)

@event.listens_for(sync_maker, "before_commit")
def before_commit(session):
    print("before commit")
async def main():
    async_session = maker()
    await async_session.commit()
asyncio.run(main())
```
Output:
```
before commit
```

asyncio and events, two opposites

SQLAlchemy events by their nature take place within the interior of a particular SQLAlchemy process; that is, an event always occurs after some particular SQLAlchemy API has been invoked by end-user code, and before some other internal aspect of that API occurs.

Contrast this to the architecture of the asyncio extension, which takes place on the exterior of SQLAlchemy’s usual flow from end-user API to DBAPI function.

The flow of messaging may be visualized as follows:

 SQLAlchemy    SQLAlchemy        SQLAlchemy          SQLAlchemy   plain
  asyncio      asyncio           ORM/Core            asyncio      asyncio
  (public      (internal)                            (internal)
  facing)
-------------|------------|------------------------|-----------|------------
asyncio API  |            |                        |           |
call  ->     |            |                        |           |
             |  ->  ->    |                        |  ->  ->   |
             |~~~~~~~~~~~~| sync API call ->       |~~~~~~~~~~~|
             | asyncio    |  event hooks ->        | sync      |
             | to         |   invoke action ->     | to        |
             | sync       |    event hooks ->      | asyncio   |
             | (greenlet) |     dialect ->         | (leave    |
             |~~~~~~~~~~~~|      event hooks ->    | greenlet) |
             |  ->  ->    |       sync adapted     |~~~~~~~~~~~|
             |            |               DBAPI -> |  ->  ->   | asyncio
             |            |                        |           | driver -> database

Where above, an API call always starts as asyncio, flows through the synchronous API, and ends as asyncio, before results are propagated through this same chain in the opposite direction. In between, the message is adapted first into sync-style API use, and then back out to async style. Event hooks then by their nature occur in the middle of the “sync-style API use”. From this it follows that the API presented within event hooks occurs inside the process by which asyncio API requests have been adapted to sync, and outgoing messages to the database API will be converted to asyncio transparently.

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 class provides a method AdaptedConnection.run_async() that allows an awaitable function to be invoked within the “synchronous” context of an event handler or other SQLAlchemy internal. This method is directly analogous to the 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 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 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 with different event loops when using the default pool implementation.

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

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

Using asyncio scoped session

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

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 the AsyncSession directly to the awaitable functions that need it.

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

from asyncio import current_task
from sqlalchemy.ext.asyncio import (
    async_scoped_session,
    async_sessionmaker,
)
async_session_factory = async_sessionmaker(
    some_async_engine,
    expire_on_commit=False,
)
AsyncScopedSession = async_scoped_session(
    async_session_factory,
    scopefunc=current_task,
)
some_async_session = AsyncScopedSession()

Warning

The “scopefunc” used by async_scoped_session is invoked an arbitrary number of times within a task, once for each time the underlying 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 will remain in memory, essentially creating a memory leak. See the following example which illustrates the correct use of async_scoped_session.remove().

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

async def some_function(some_async_session, some_object):
    # use the AsyncSession directly
    some_async_session.add(some_object)
    # use the AsyncSession via the context-local proxy
    await AsyncScopedSession.commit()
    # "remove" the current proxied AsyncSession for the local context
    await AsyncScopedSession.remove()

New in version 1.4.19.

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

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

function sqlalchemy.ext.asyncio.create_async_engine(url: Union[str, ], **kw: Any) → AsyncEngine

Create a new async engine instance.

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

New in version 1.4.

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

New in version 1.4.29.

class sqlalchemy.ext.asyncio.AsyncEngine

An asyncio proxy for a Engine.

is acquired using the create_async_engine() function:

from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://user:pass@host/dbname")

New in version 1.4.

Members

, clear_compiled_cache(), , dialect, , driver, , engine, , get_execution_options(), , pool, , sync_engine, , url

Class signature

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

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

Return a context manager which when entered will deliver an AsyncConnection with an 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 class on behalf of the AsyncEngine class.

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

New in version 1.4.
method sqlalchemy.ext.asyncio.AsyncEngine.connect() →

Return an AsyncConnection object.

The 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 method.
attribute sqlalchemy.ext.asyncio.AsyncEngine.dialect

Proxy for the Engine.dialect attribute on behalf of the class.
method sqlalchemy.ext.asyncio.AsyncEngine.async dispose(close: bool = True) → None

Dispose of the connection pool used by this .
- Parameters:
  
  close –
  
  if left at its default of True, has the effect of fully closing all currently checked in database connections. Connections that are still checked out will not be closed, however they will no longer be associated with this Engine, so when they are closed individually, eventually the which they are associated with will be garbage collected and they will be closed out fully, if not already closed on checkin.
  
  If set to False, the previous connection pool is de-referenced, and otherwise not touched in any way.
See also

Engine.dispose()
attribute driver

Driver name of the Dialect in use by this Engine.

Proxied for the class on behalf of the AsyncEngine class.
attribute echo

When True, enable log output for this element.

Proxied for the Engine class on behalf of the class.

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

Returns this .

Proxied for the Engine class on behalf of the class.

Used for legacy schemes that accept Connection / objects within the same variable.
method sqlalchemy.ext.asyncio.AsyncEngine.execution_options(**opt: Any) →

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

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

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

Proxied for the Engine class on behalf of the class.

See also

Engine.execution_options()
attribute name

String name of the Dialect in use by this Engine.

Proxied for the class on behalf of the AsyncEngine class.
attribute pool

Proxy for the Engine.pool attribute on behalf of the AsyncEngine class.
method async raw_connection() → PoolProxiedConnection

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

See also
attribute sqlalchemy.ext.asyncio.AsyncEngine.sync_engine:

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

This instance can be used as an event target.

See also

Using events with the asyncio extension
method update_execution_options(**opt: Any) → None

Update the default execution_options dictionary of this Engine.

Proxied for the 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 .

See also

Connection.execution_options()
attribute sqlalchemy.ext.asyncio.AsyncEngine.url

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

class sqlalchemy.ext.asyncio.AsyncConnection

An asyncio proxy for a Connection.

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

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

begin(), , close(), , commit(), , default_isolation_level, , exec_driver_sql(), , execution_options(), , get_raw_connection(), , in_nested_transaction(), , info, , invalidated, , run_sync(), , scalars(), , stream(), , sync_connection,

Class signature

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

method begin() → AsyncTransaction

Begin a transaction prior to autobegin occurring.
method begin_nested() → AsyncTransaction

Begin a nested transaction and return a transaction handle.
method async close() → None

Close this AsyncConnection.

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

Return True if this connection is closed.

Proxied for the Connection class on behalf of the 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 automatically whenever a statement is first executed, or when the Connection.begin() method is called.
attribute connection

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

The default isolation level assigned to this Connection.

Proxied for the class on behalf of the AsyncConnection class.

This is the isolation level setting that the has when first procured via the Engine.connect() method. This level stays in place until the is used to change the setting on a per-Connection basis.

Unlike , this attribute is set ahead of time from the first connection procured by the dialect, so SQL query is not invoked when this accessor is called.

New in version 0.9.9.

See also

Connection.get_isolation_level() - view current level

- set per Engine isolation level

- set per Connection isolation level
attribute dialect

Proxy for the Connection.dialect attribute on behalf of the AsyncConnection class.
method async exec_driver_sql(statement: str, parameters: Optional[_DBAPIAnyExecuteParams] = None, execution_options: Optional[CoreExecuteOptionsParameter] = None) → CursorResult[Any]

Executes a driver-level SQL string and return buffered .
method sqlalchemy.ext.asyncio.AsyncConnection.async execute(statement: , parameters: Optional[_CoreAnyExecuteParams] = None, *, execution_options: Optional[CoreExecuteOptionsParameter] = None) → CursorResult[Any]

Executes a SQL statement construct and return a buffered .
- Parameters:
  - object –
    
    The statement to be executed. This is always an object that is in both the ClauseElement and hierarchies, including:
    - Select
    - , Update,
    - TextClause and
    - DDL and objects which inherit from
  - 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 object.
method sqlalchemy.ext.asyncio.AsyncConnection.async execution_options(**opt: Any) →

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

This returns this AsyncConnection object with the new options added.

See for full details on this method.
method sqlalchemy.ext.asyncio.AsyncConnection.get_nested_transaction() → Optional[]

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

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

New in version 1.4.0b2.
method sqlalchemy.ext.asyncio.AsyncConnection.async get_raw_connection() →

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

This is a SQLAlchemy connection-pool proxied connection which then has the attribute _ConnectionFairy.driver_connection that refers to the actual driver connection. Its _ConnectionFairy.dbapi_connection refers instead to an instance that adapts the driver connection to the DBAPI protocol.
method sqlalchemy.ext.asyncio.AsyncConnection.get_transaction() → Optional[]

Return an AsyncTransaction representing the current transaction, if any.

This makes use of the underlying synchronous connection’s method to get the current Transaction, which is then proxied in a new 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 in_transaction() → bool

Return True if a transaction is in progress.
attribute sqlalchemy.ext.asyncio.AsyncConnection.info

Return the 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 is currently connected. If the AsyncConnection.closed attribute is True, then accessing this attribute will raise .

New in version 1.4.0b2.
method sqlalchemy.ext.asyncio.AsyncConnection.async invalidate(exception: Optional[BaseException] = None) → None

Invalidate the underlying DBAPI connection associated with this .

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

Return True if this connection was invalidated.

Proxied for the Connection class on behalf of the 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 automatically whenever a statement is first executed, or when the Connection.begin() method is called.
method async run_sync(fn: Callable[[…], Any], *arg: Any, **kw: Any) → Any

Invoke the given sync callable passing self as the first argument.

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.

E.g.:
```
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.

See also

Running Synchronous Methods and Functions under asyncio
method async scalar(statement: Executable, parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[CoreExecuteOptionsParameter] = None) → Any

Executes a SQL statement construct and returns a scalar object.

This method is shorthand for invoking the method after invoking the Connection.execute() method. Parameters are equivalent.
- Returns:
  
  a scalar Python value representing the first column of the first row returned.
method async scalars(statement: Executable, parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[CoreExecuteOptionsParameter] = None) → [Any]

Executes a SQL statement construct and returns a scalar objects.

This method is shorthand for invoking the Result.scalars() method after invoking the method. Parameters are equivalent.
- Returns:
  
  a ScalarResult object.
New in version 1.4.24.
method async start(is_ctxmanager: bool = False) → AsyncConnection

Start this object’s context outside of using a Python with: block.
method sqlalchemy.ext.asyncio.AsyncConnection.stream(statement: , parameters: Optional[_CoreAnyExecuteParams] = None, *, execution_options: Optional[CoreExecuteOptionsParameter] = None) → AsyncIterator[AsyncResult[Any]]

Execute a statement and return an awaitable yielding a object.

E.g.:
```
result = await conn.stream(stmt):
async for row in result:
    print(f"{row}")
```
The AsyncConnection.stream() method supports optional context manager use against the object, as in:
```
async with conn.stream(stmt) as result:
    async for row in result:
        print(f"{row}")
```
In the above pattern, the AsyncResult.close() method is invoked unconditionally, even if the iterator is interrupted by an exception throw. Context manager use remains optional, however, and the function may be called in either an async with fn(): or await fn() style.

New in version 2.0.0b3: added context manager support
- Returns:
  
  an awaitable object that will yield an object.
See also

AsyncConnection.stream_scalars()
method stream_scalars(statement: Executable, parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[CoreExecuteOptionsParameter] = None) → AsyncIterator[[Any]]

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

E.g.:
```
result = await conn.stream_scalars(stmt):
async for scalar in result:
    print(f"{scalar}")
```
This method is shorthand for invoking the AsyncResult.scalars() method after invoking the Connection.stream() method. Parameters are equivalent.

The method supports optional context manager use against the AsyncScalarResult object, as in:
```
async with conn.stream_scalars(stmt) as result:
    async for scalar in result:
        print(f"{scalar}")
```
In the above pattern, the method is invoked unconditionally, even if the iterator is interrupted by an exception throw. Context manager use remains optional, however, and the function may be called in either an async with fn(): or style.

New in version 2.0.0b3: added context manager support
- Returns:
  
  an awaitable object that will yield an AsyncScalarResult object.
New in version 1.4.24.

See also
attribute sqlalchemy.ext.asyncio.AsyncConnection.sync_connection: Optional[]

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

This instance can be used as an event target.

See also

Using events with the asyncio extension
attribute sync_engine: Engine

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

This instance can be used as an event target.

See also

Using events with the asyncio extension

class sqlalchemy.ext.asyncio.AsyncTransaction

An asyncio proxy for a .

Members

close(), , rollback(),

Class signature

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

method 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 async commit() → None

Commit this AsyncTransaction.
method async rollback() → None

Roll back this AsyncTransaction.
method async start(is_ctxmanager: bool = False) → AsyncTransaction

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

Result Set API Documentation

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

class sqlalchemy.ext.asyncio.AsyncResult

An asyncio wrapper around a object.

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

Note

As is the case with , 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 object. For in-Python de-duplication of instances or rows, use the AsyncResult.unique() modifier method.

New in version 1.4.

Members

, close(), , columns(), , fetchmany(), , first(), , keys(), , one(), , partitions(), , scalar_one(), , scalars(), , tuples(), , yield_per()

Class signature

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

method sqlalchemy.ext.asyncio.AsyncResult.async all() → Sequence[[_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 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 columns(*col_expressions: _KeyIndexType) → SelfAsyncResult

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 async fetchall() → Sequence[Row[_TP]]

A synonym for the method.

New in version 2.0.
method sqlalchemy.ext.asyncio.AsyncResult.async fetchmany(size: Optional[int] = None) → Sequence[[_TP]]

Fetch many rows.

When all rows are exhausted, returns an empty list.

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

To fetch rows in groups, use the AsyncResult.partitions() method.
- Returns:
  
  a list of objects.
See also

AsyncResult.partitions()
method async fetchone() → Optional[Row[_TP]]

Fetch one row.

When all rows are exhausted, returns None.

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

To fetch the first row of a result only, use the method. To iterate through all rows, iterate the AsyncResult object directly.
- a object if no filters are applied, or None if no rows remain.
method sqlalchemy.ext.asyncio.AsyncResult.async first() → Optional[[_TP]]

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

Closes the result set and discards remaining rows.

Note

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

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

See also
- Returns:
  
  a Row object, or None if no rows remain.
See also

AsyncResult.one()
method async freeze() → FrozenResult[_TP]

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

The callable object returned is an instance of FrozenResult.

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

See also

- example usage within the ORM to implement a result-set cache.
method sqlalchemy.ext.asyncio.AsyncResult.keys() → RMKeyView

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

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

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

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

Changed in version 1.4: a key view object is returned rather than a plain list.
method sqlalchemy.ext.asyncio.AsyncResult.mappings() →

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

When this filter is applied, fetching rows will return objects instead of Row objects.
- Returns:
  
  a new filtering object referring to the underlying Result object.
method async one() → Row[_TP]

Return exactly one row or raise an exception.

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

Note

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

New in version 1.4.
- Returns:
  
  The first Row.
  
  Raises:
  
  , NoResultFound
See also

AsyncResult.one_or_none()
method sqlalchemy.ext.asyncio.AsyncResult.async one_or_none() → Optional[[_TP]]

Return at most one result or raise an exception.

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

New in version 1.4.
- Returns:
  
  The first or None if no row is available.
  
  Raises:
  
  MultipleResultsFound
See also

AsyncResult.one()
method async partitions(size: Optional[int] = 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 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 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 and then AsyncResult.one().

See also

AsyncResult.scalars()
method async scalar_one_or_none() → Optional[Any]

Return exactly one scalar result or None.

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

See also

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

Return an AsyncScalarResult filtering object which will return single elements rather than 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 filtering object referring to this AsyncResult object.
attribute t

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

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

New in version 2.0.
method sqlalchemy.ext.asyncio.AsyncResult.tuples() → [_TP]

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

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

New in version 2.0.
- Returns:
  
  the AsyncTupleResult type at typing time.
See also

AsyncResult.t - shorter synonym

- Row version
method yield_per(num: int) → SelfFilterResult

inherited from the FilterResult.yield_per() method of

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

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

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

See also

- describes Core behavior for Result.yield_per()

- in the ORM Querying Guide

class sqlalchemy.ext.asyncio.AsyncScalarResult

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

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

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

New in version 1.4.

Members

all(), , closed, , fetchmany(), , one(), , partitions(), , yield_per()

Class signature

class (sqlalchemy.ext.asyncio.AsyncCommon)

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

Return all scalar values in a list.

Equivalent to except that scalar values, rather than Row objects, are returned.
method async close() → None

inherited from the AsyncCommon.close() method of AsyncCommon

Close this result.
attribute sqlalchemy.ext.asyncio.AsyncScalarResult.closed

inherited from the AsyncCommon.closed attribute of AsyncCommon

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

New in version 2.0.0b3.
method async fetchall() → Sequence[_R]

A synonym for the AsyncScalarResult.all() method.
method async fetchmany(size: Optional[int] = None) → Sequence[_R]

Fetch many objects.

Equivalent to AsyncResult.fetchmany() except that scalar values, rather than objects, are returned.
method sqlalchemy.ext.asyncio.AsyncScalarResult.async first() → Optional[_R]

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

Equivalent to except that scalar values, rather than Row objects, are returned.
method async one() → _R

Return exactly one object or raise an exception.

Equivalent to AsyncResult.one() except that scalar values, rather than objects, are returned.
method sqlalchemy.ext.asyncio.AsyncScalarResult.async one_or_none() → Optional[_R]

Return at most one object or raise an exception.

Equivalent to except that scalar values, rather than Row objects, are returned.
method async partitions(size: Optional[int] = None) → AsyncIterator[Sequence[_R]]

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

Equivalent to AsyncResult.partitions() except that scalar values, rather than objects, are returned.
method sqlalchemy.ext.asyncio.AsyncScalarResult.unique(strategy: Optional[_UniqueFilterType] = None) → SelfAsyncScalarResult

Apply unique filtering to the objects returned by this .

See AsyncResult.unique() for usage details.
method yield_per(num: int) → SelfFilterResult

inherited from the FilterResult.yield_per() method of

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

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

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

See also

- describes Core behavior for Result.yield_per()

- in the ORM Querying Guide

class sqlalchemy.ext.asyncio.AsyncMappingResult

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

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

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

New in version 1.4.

Members

all(), , closed, , fetchall(), , fetchone(), , keys(), , one_or_none(), , unique(),

Class signature

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

method async all() → Sequence[RowMapping]

Return all rows in a list.

Equivalent to except that RowMapping values, rather than objects, are returned.
method sqlalchemy.ext.asyncio.AsyncMappingResult.async close() → None

inherited from the AsyncCommon.close() method of AsyncCommon

Close this result.
attribute closed

inherited from the AsyncCommon.closed attribute of AsyncCommon

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

New in version 2.0.0b3.
method sqlalchemy.ext.asyncio.AsyncMappingResult.columns(*col_expressions: _KeyIndexType) → SelfAsyncMappingResult

Establish the columns that should be returned in each row.
method async fetchall() → Sequence[RowMapping]

A synonym for the method.
method sqlalchemy.ext.asyncio.AsyncMappingResult.async fetchmany(size: Optional[int] = None) → Sequence[]

Fetch many rows.

Equivalent to AsyncResult.fetchmany() except that values, rather than Row objects, are returned.
method async fetchone() → Optional[RowMapping]

Fetch one object.

Equivalent to except that RowMapping values, rather than objects, are returned.
method sqlalchemy.ext.asyncio.AsyncMappingResult.async first() → Optional[]

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

Equivalent to AsyncResult.first() except that values, rather than Row objects, are returned.
method keys() → RMKeyView

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

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

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

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

Changed in version 1.4: a key view object is returned rather than a plain list.
method async one() → RowMapping

Return exactly one object or raise an exception.

Equivalent to except that RowMapping values, rather than objects, are returned.
method sqlalchemy.ext.asyncio.AsyncMappingResult.async one_or_none() → Optional[]

Return at most one object or raise an exception.

Equivalent to AsyncResult.one_or_none() except that values, rather than Row objects, are returned.
method async partitions(size: Optional[int] = None) → AsyncIterator[Sequence[RowMapping]]

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

Equivalent to except that RowMapping values, rather than objects, are returned.
method sqlalchemy.ext.asyncio.AsyncMappingResult.unique(strategy: Optional[_UniqueFilterType] = None) → SelfAsyncMappingResult

Apply unique filtering to the objects returned by this .

See AsyncResult.unique() for usage details.
method yield_per(num: int) → SelfFilterResult

inherited from the FilterResult.yield_per() method of

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

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

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

See also

- describes Core behavior for Result.yield_per()

- in the ORM Querying Guide

class sqlalchemy.ext.asyncio.AsyncTupleResult

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

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

Class signature

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

function sqlalchemy.ext.asyncio.async_object_session(instance: object) → Optional[AsyncSession]

Return the to which the given instance belongs.

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

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

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

Parameters:

instance – an ORM mapped instance

Returns:

an object, or None.

New in version 1.4.18.

function sqlalchemy.ext.asyncio.async_session(session: Session) → Optional[]

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

Parameters:

session – a Session instance.

Returns:

a instance, or None.

New in version 1.4.18.

class sqlalchemy.ext.asyncio.async_sessionmaker

A configurable AsyncSession factory.

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

e.g.:

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker
async def run_some_sql(async_session: async_sessionmaker[AsyncSession]) -> None:
    async with async_session() as session:
        session.add(SomeObject(data="object"))
        session.add(SomeOtherObject(name="other object"))
        await session.commit()
async def main() -> None:
    # an AsyncEngine, which the AsyncSession will use for connection
    # resources
    engine = create_async_engine('postgresql+asyncpg://scott:tiger@localhost/')
    # create a reusable factory for new AsyncSession instances
    async_session = async_sessionmaker(engine)
    await run_some_sql(async_session)
    await engine.dispose()

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

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

See also

- shows example use

sessionmaker - general overview of the

architecture

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

Members

__call__(), , begin(),

Class signature

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

method __call__(**local_kw: Any) → _AS

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

In Python, the __call__ method is invoked on an object when it is “called” in the same way as a function:
method sqlalchemy.ext.asyncio.async_sessionmaker.__init__(bind: Optional[_AsyncSessionBind] = None, *, class\: Type[_AS] = <class ‘sqlalchemy.ext.asyncio.session.AsyncSession’>, _autoflush: bool = True, expire_on_commit: bool = True, info: Optional[_InfoType] = None, **kw: Any)

Construct a new .

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

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

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

e.g.:

async def main():
    Session = async_sessionmaker(some_engine)
    async with Session.begin() as session:
        session.add(some_object)
    # commits transaction, closes session

method configure(**new_kw: Any) → None

(Re)configure the arguments for this async_sessionmaker.

e.g.:

AsyncSession = async_sessionmaker(some_engine)
AsyncSession.configure(bind=create_async_engine('sqlite+aiosqlite://'))

class sqlalchemy.ext.asyncio.async_scoped_session

Provides scoped management of AsyncSession objects.

See the section for usage details.

New in version 1.4.19.

Members

__call__(), , add(), , autoflush, , begin_nested(), , close(), , commit(), , connection(), , deleted, , execute(), , expire_all(), , expunge_all(), , get(), , identity_key(), , info, , is_active, , merge(), , no_autoflush, , refresh(), , rollback(), , scalars(), , stream(),

Class signature

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

method __call__(**kw: Any) → _AS

Return the current AsyncSession, creating it using the if not present.
- Parameters:
  
  **kw – Keyword arguments will be passed to the scoped_session.session_factory callable, if an existing is not present. If the AsyncSession is present and keyword arguments have been passed, is raised.
method sqlalchemy.ext.asyncio.async_scoped_session.__init__(session_factory: [_AS], scopefunc: Callable[[], Any])

Construct a new async_scoped_session.
- Parameters:
  - session_factory – a factory to create new instances. This is usually, but not necessarily, an instance of async_sessionmaker.
  - scopefunc – function which defines the current scope. A function such as asyncio.current_task may be useful here.

method add(instance: object, _warn: bool = True) → None

Place an object into this Session.

Proxied for the class on behalf of the async_scoped_session class.

Proxied for the class on behalf of the AsyncSession class.

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

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

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

See also

Session.add_all()

- at Basics of Using a Session
method add_all(instances: Iterable[object]) → None

Add the given collection of instances to this Session.

Proxied for the class on behalf of the async_scoped_session class.

Proxied for the class on behalf of the AsyncSession class.

See the documentation for for a general behavioral description.

See also

Session.add()

- at Basics of Using a Session
attribute autoflush

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

Proxied for the class on behalf of the async_scoped_session class.
method begin() → AsyncSessionTransaction

Return an object.

Proxied for the AsyncSession class on behalf of the class.

The underlying Session will perform the “begin” action when the object is entered:
```
async with async_session.begin():
    # .. ORM transaction is begun
```
Note that database IO will not normally occur when the session-level transaction is begun, as database transactions begin on an on-demand basis. However, the begin block is async to accommodate for a SessionEvents.after_transaction_create() event hook that may perform IO.

For a general description of ORM begin, see .
method sqlalchemy.ext.asyncio.async_scoped_session.begin_nested() →

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

Proxied for the class on behalf of the async_scoped_session class.

Behavior is the same as that of .

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

Proxy for the AsyncSession.bind attribute on behalf of the async_scoped_session class.
method async close() → None

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

Proxied for the class on behalf of the async_scoped_session class.

This expunges all ORM objects associated with this , ends any transaction in progress and releases any objects which this AsyncSession itself has checked out from associated objects. The operation then leaves the AsyncSession in a state which it may be used again.

Tip

The method does not prevent the Session from being used again. The AsyncSession itself does not actually have a distinct “closed” state; it merely means the will release all database connections and ORM objects.

See also

Closing - detail on the semantics of
async classmethod sqlalchemy.ext.asyncio.async_scoped_session.close_all() → None

Close all sessions.

Proxied for the AsyncSession class on behalf of the class.
method sqlalchemy.ext.asyncio.async_scoped_session.async commit() → None

Commit the current transaction in progress.

Proxied for the class on behalf of the async_scoped_session class.
method configure(**kwargs: Any) → None

reconfigure the sessionmaker used by this .

See sessionmaker.configure().
method async connection(**kw: Any) → AsyncConnection

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

Proxied for the class on behalf of the async_scoped_session class.

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

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

See also

Session.connection() - main documentation for “connection”
method async delete(instance: object) → None

Mark an instance as deleted.

Proxied for the AsyncSession class on behalf of the class.

The database delete operation occurs upon flush().

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

See also

Session.delete() - main documentation for delete
attribute deleted

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

Proxied for the AsyncSession class on behalf of the class.

Proxied for the Session class on behalf of the class.
attribute sqlalchemy.ext.asyncio.async_scoped_session.dirty

The set of all persistent instances considered dirty.

Proxied for the class on behalf of the async_scoped_session class.

Proxied for the class on behalf of the AsyncSession class.

E.g.:
```
some_mapped_object in session.dirty
```
Instances are considered dirty when they were modified but not deleted.

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

To check if an instance has actionable net changes to its attributes, use the method.
method sqlalchemy.ext.asyncio.async_scoped_session.async execute(statement: , params: Optional[_CoreAnyExecuteParams] = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: Optional[_BindArguments] = None, **kw: Any) → Result[Any]

Execute a statement and return a buffered object.

Proxied for the AsyncSession class on behalf of the class.

See also

Session.execute() - main documentation for execute
method expire(instance: object, attribute_names: Optional[Iterable[str]] = None) → None

Expire the attributes on an instance.

Proxied for the AsyncSession class on behalf of the class.

Proxied for the Session class on behalf of the class.

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

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

The Session object’s default behavior is to expire all state whenever the Session.rollback() or Session.commit() methods are called, so that new state can be loaded for the new transaction. For this reason, calling Session.expire() only makes sense for the specific case that a non-ORM SQL statement was emitted in the current transaction.
- Parameters:
  - instance – The instance to be refreshed.
  - attribute_names – optional list of string attribute names indicating a subset of attributes to be expired.

See also
[Refreshing / Expiring]($592600365cded3ff.md#session-expire) - introductory material
[Session.expire()]($694f628462946390.md#sqlalchemy.orm.Session.expire "sqlalchemy.orm.Session.expire")
[Session.refresh()]($694f628462946390.md#sqlalchemy.orm.Session.refresh "sqlalchemy.orm.Session.refresh")
[Query.populate\_existing()]($3d0cc000ec6c7150.md#sqlalchemy.orm.Query.populate_existing "sqlalchemy.orm.Query.populate_existing")

method expire_all() → None

Expires all persistent instances within this Session.

Proxied for the AsyncSession class on behalf of the class.

Proxied for the Session class on behalf of the class.

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

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

See also

- introductory material

Session.expire()

Query.populate_existing()
method expunge(instance: object) → None

Remove the instance from this Session.

Proxied for the AsyncSession class on behalf of the class.

Proxied for the Session class on behalf of the class.

This will free all internal references to the instance. Cascading will be applied according to the expunge cascade rule.
method sqlalchemy.ext.asyncio.async_scoped_session.expunge_all() → None

Remove all object instances from this Session.

Proxied for the class on behalf of the async_scoped_session class.

Proxied for the class on behalf of the AsyncSession class.

This is equivalent to calling expunge(obj) on all objects in this Session.
method async flush(objects: Optional[Sequence[Any]] = None) → None

Flush all the object changes to the database.

Proxied for the AsyncSession class on behalf of the class.

See also

Session.flush() - main documentation for flush
method async get(entity: _EntityBindKey[_O], ident: _PKIdentityArgument, *, options: Optional[Sequence[ORMOption]] = None, populate_existing: bool = False, with_for_update: Optional[ForUpdateArg] = None, identity_token: Optional[Any] = None, execution_options: OrmExecuteOptionsParameter = {}) → Optional[_O]

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

Proxied for the class on behalf of the async_scoped_session class.

See also

- main documentation for get
method sqlalchemy.ext.asyncio.async_scoped_session.get_bind(mapper: Optional[_EntityBindKey[_O]] = None, clause: Optional[] = None, bind: Optional[_SessionBind] = None, **kw: Any) → Union[Engine, ]

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

Proxied for the class on behalf of the async_scoped_session class.

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

Note

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

The pattern introduced at Custom Vertical Partitioning illustrates how to apply a custom bind-lookup scheme to a given a set of Engine objects. To apply a corresponding implementation for use with a AsyncSession and objects, continue to subclass Session and apply it to using AsyncSession.sync_session_class. The inner method must continue to return instances, which can be acquired from a AsyncEngine using the attribute:

```

using example from “Custom Vertical Partitioning”

import random
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.orm import Session
# construct async engines w/ async drivers
engines = {
    'leader':create_async_engine("sqlite+aiosqlite:///leader.db"),
    'other':create_async_engine("sqlite+aiosqlite:///other.db"),
    'follower1':create_async_engine("sqlite+aiosqlite:///follower1.db"),
    'follower2':create_async_engine("sqlite+aiosqlite:///follower2.db"),
}
class RoutingSession(Session):
    def get_bind(self, mapper=None, clause=None, **kw):
        # within get_bind(), return sync engines
        if mapper and issubclass(mapper.class_, MyOtherClass):
            return engines['other'].sync_engine
        elif self._flushing or isinstance(clause, (Update, Delete)):
            return engines['leader'].sync_engine
        else:
            return engines[
                random.choice(['follower1','follower2'])
            ].sync_engine
# apply to AsyncSession using sync_session_class
AsyncSessionMaker = async_sessionmaker(
    sync_session_class=RoutingSession
)
```
The [Session.get\_bind()]($694f628462946390.md#sqlalchemy.orm.Session.get_bind "sqlalchemy.orm.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()](#sqlalchemy.ext.asyncio.AsyncSession.run_sync "sqlalchemy.ext.asyncio.AsyncSession.run_sync"), so routines that wish to run SQL commands inside of [Session.get\_bind()]($694f628462946390.md#sqlalchemy.orm.Session.get_bind "sqlalchemy.orm.Session.get_bind") can continue to do so using blocking-style code, which will be translated to implicitly async calls at the point of invoking IO on the database drivers.

classmethod identity_key(class\: Optional[Type[Any]] = None, _ident: Union[Any, Tuple[Any, …]] = None, *, instance: Optional[Any] = None, row: Optional[Union[[Any], RowMapping]] = None, identity_token: Optional[Any] = None) → _IdentityKeyType[Any]

Return an identity key.

Proxied for the class on behalf of the async_scoped_session class.

Proxied for the class on behalf of the AsyncSession class.

This is an alias of .
attribute sqlalchemy.ext.asyncio.async_scoped_session.identity_map

Proxy for the attribute on behalf of the AsyncSession class.

Proxied for the class on behalf of the async_scoped_session class.
attribute info

A user-modifiable dictionary.

Proxied for the AsyncSession class on behalf of the class.

Proxied for the Session class on behalf of the class.

The initial value of this dictionary can be populated using the info argument to the Session constructor or constructor or factory methods. The dictionary here is always local to this Session and can be modified independently of all other objects.
method sqlalchemy.ext.asyncio.async_scoped_session.async invalidate() → None

Close this Session, using connection invalidation.

Proxied for the class on behalf of the async_scoped_session class.

For a complete description, see .
attribute sqlalchemy.ext.asyncio.async_scoped_session.is_active

True if this not in “partial rollback” state.

Proxied for the AsyncSession class on behalf of the class.

Proxied for the Session class on behalf of the class.

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

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

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

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

See also

Session.in_transaction()
method 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 class.

Proxied for the Session class on behalf of the class.

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

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

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

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

Proxied for the class on behalf of the async_scoped_session class.

See also

- main documentation for merge
attribute sqlalchemy.ext.asyncio.async_scoped_session.new

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

Proxied for the class on behalf of the async_scoped_session class.

Proxied for the class on behalf of the AsyncSession class.
attribute no_autoflush

Return a context manager that disables autoflush.

Proxied for the AsyncSession class on behalf of the class.

Proxied for the Session class on behalf of the 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) → Optional[]

Return the Session to which an object belongs.

Proxied for the class on behalf of the async_scoped_session class.

Proxied for the class on behalf of the AsyncSession class.

This is an alias of .
method sqlalchemy.ext.asyncio.async_scoped_session.async refresh(instance: object, attribute_names: Optional[Iterable[str]] = None, with_for_update: Optional[ForUpdateArg] = None) → None

Expire and refresh the attributes on the given instance.

Proxied for the 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 method. See that method for a complete description of all options.

See also

Session.refresh() - main documentation for refresh
method 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 async rollback() → None

Rollback the current transaction in progress.

Proxied for the AsyncSession class on behalf of the class.
method sqlalchemy.ext.asyncio.async_scoped_session.async scalar(statement: , params: Optional[_CoreSingleExecuteParams] = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: Optional[_BindArguments] = None, **kw: Any) → Any

Execute a statement and return a scalar result.

Proxied for the AsyncSession class on behalf of the class.

See also

Session.scalar() - main documentation for scalar
method async scalars(statement: Executable, params: Optional[_CoreSingleExecuteParams] = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: Optional[_BindArguments] = None, **kw: Any) → [Any]

Execute a statement and return scalar results.

Proxied for the AsyncSession class on behalf of the class.
- Returns:
  
  a ScalarResult object
New in version 1.4.24.

See also

- main documentation for scalars

AsyncSession.stream_scalars() - streaming version
attribute 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 is needed.
method sqlalchemy.ext.asyncio.async_scoped_session.async stream(statement: , params: Optional[_CoreAnyExecuteParams] = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: Optional[_BindArguments] = None, **kw: Any) → AsyncResult[Any]

Execute a statement and return a streaming object.

Proxied for the AsyncSession class on behalf of the class.
method sqlalchemy.ext.asyncio.async_scoped_session.async stream_scalars(statement: , params: Optional[_CoreSingleExecuteParams] = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: Optional[_BindArguments] = None, **kw: Any) → AsyncScalarResult[Any]

Execute a statement and return a stream of scalar results.

Proxied for the class on behalf of the async_scoped_session class.
- Returns:
  
  an object
New in version 1.4.24.

See also

Session.scalars() - main documentation for scalars

- non streaming version

class sqlalchemy.ext.asyncio.AsyncSession

Asyncio version of Session.

The is a proxy for a traditional Session instance.

New in version 1.4.

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

Members

sync_session_class, , add(), , autoflush, , begin_nested(), , close_all(), , connection(), , deleted, , execute(), , expire_all(), , expunge_all(), , get(), , get_nested_transaction(), , identity_key(), , in_nested_transaction(), , info, , is_active, , merge(), , no_autoflush, , refresh(), , run_sync(), , scalars(), , stream_scalars(),

Class signature

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

attribute sync_session_class: Type[Session] = <class ‘sqlalchemy.orm.session.Session’>

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

At the class level, this attribute is the default value for the 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 instance for this AsyncSession instance.

New in version 1.4.24.
method __init__(bind: Optional[_AsyncSessionBind] = None, *, binds: Optional[Dict[_SessionBindKey, _AsyncSessionBind]] = None, sync_session_class: Optional[Type[Session]] = None, **kw: Any)

Construct a new .

All parameters other than sync_session_class are passed to the sync_session_class callable directly to instantiate a new Session. Refer to for parameter documentation.
- Parameters:
  
  sync_session_class –
  
  A Session subclass or other callable which will be used to construct the which will be proxied. This parameter may be used to provide custom Session subclasses. Defaults to the class-level attribute.
  
  New in version 1.4.24.
method sqlalchemy.ext.asyncio.AsyncSession.add(instance: object, _warn: bool = True) → None

Place an object into this .

Proxied for the Session class on behalf of the class.

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

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

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

See also

Adding New or Existing Items - at
method sqlalchemy.ext.asyncio.AsyncSession.add_all(instances: Iterable[object]) → None

Add the given collection of instances to this .

Proxied for the Session class on behalf of the class.

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

See also

Adding New or Existing Items - at
attribute sqlalchemy.ext.asyncio.AsyncSession.autoflush

Proxy for the Session.autoflush attribute on behalf of the class.
method sqlalchemy.ext.asyncio.AsyncSession.begin() →

Return an AsyncSessionTransaction object.

The underlying will perform the “begin” action when the AsyncSessionTransaction object is entered:
```
async with async_session.begin():
    # .. ORM transaction is begun
```
Note that database IO will not normally occur when the session-level transaction is begun, as database transactions begin on an on-demand basis. However, the begin block is async to accommodate for a event hook that may perform IO.

For a general description of ORM begin, see Session.begin().
method begin_nested() → AsyncSessionTransaction

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

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

For a general description of ORM begin nested, see .
method sqlalchemy.ext.asyncio.AsyncSession.async close() → None

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

This expunges all ORM objects associated with this AsyncSession, ends any transaction in progress and any AsyncConnection objects which this itself has checked out from associated AsyncEngine objects. The operation then leaves the in a state which it may be used again.

Tip

The AsyncSession.close() method does not prevent the Session from being used again. The itself does not actually have a distinct “closed” state; it merely means the AsyncSession will release all database connections and ORM objects.

See also

- detail on the semantics of AsyncSession.close()
async classmethod close_all() → None

Close all AsyncSession sessions.
method async commit() → None

Commit the current transaction in progress.
method sqlalchemy.ext.asyncio.AsyncSession.async connection(**kw: Any) →

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

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

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

See also

- main documentation for “connection”
method sqlalchemy.ext.asyncio.AsyncSession.async delete(instance: object) → None

Mark an instance as deleted.

The database delete operation occurs upon flush().

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

See also

- main documentation for delete
attribute sqlalchemy.ext.asyncio.AsyncSession.deleted

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

Proxied for the class on behalf of the AsyncSession class.
attribute dirty

The set of all persistent instances considered dirty.

Proxied for the Session class on behalf of the class.

E.g.:
```
some_mapped_object in session.dirty
```
Instances are considered dirty when they were modified but not deleted.

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

To check if an instance has actionable net changes to its attributes, use the Session.is_modified() method.
method async execute(statement: Executable, params: Optional[_CoreAnyExecuteParams] = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: Optional[_BindArguments] = None, **kw: Any) → [Any]

Execute a statement and return a buffered Result object.

See also

- main documentation for execute
method sqlalchemy.ext.asyncio.AsyncSession.expire(instance: object, attribute_names: Optional[Iterable[str]] = None) → None

Expire the attributes on an instance.

Proxied for the class on behalf of the AsyncSession class.

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

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

The object’s default behavior is to expire all state whenever the Session.rollback() or Session.commit() methods are called, so that new state can be loaded for the new transaction. For this reason, calling Session.expire() only makes sense for the specific case that a non-ORM SQL statement was emitted in the current transaction.
- Parameters:
  - instance – The instance to be refreshed.
  - attribute_names – optional list of string attribute names indicating a subset of attributes to be expired.

See also
[Refreshing / Expiring]($592600365cded3ff.md#session-expire) - introductory material
[Session.expire()]($694f628462946390.md#sqlalchemy.orm.Session.expire "sqlalchemy.orm.Session.expire")
[Session.refresh()]($694f628462946390.md#sqlalchemy.orm.Session.refresh "sqlalchemy.orm.Session.refresh")
[Query.populate\_existing()]($3d0cc000ec6c7150.md#sqlalchemy.orm.Query.populate_existing "sqlalchemy.orm.Query.populate_existing")

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

Expires all persistent instances within this Session.

Proxied for the class on behalf of the AsyncSession class.

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

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

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

See also

- introductory material

Session.expire()

Query.populate_existing()
method expunge(instance: object) → None

Remove the instance from this Session.

Proxied for the Session class on behalf of the 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 class on behalf of the AsyncSession class.

This is equivalent to calling expunge(obj) on all objects in this Session.
method async flush(objects: Optional[Sequence[Any]] = None) → None

Flush all the object changes to the database.

See also

Session.flush() - main documentation for flush
method async get(entity: _EntityBindKey[_O], ident: _PKIdentityArgument, *, options: Optional[Sequence[ORMOption]] = None, populate_existing: bool = False, with_for_update: Optional[ForUpdateArg] = None, identity_token: Optional[Any] = None, execution_options: OrmExecuteOptionsParameter = {}) → Optional[_O]

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

See also

- main documentation for get
method sqlalchemy.ext.asyncio.AsyncSession.get_bind(mapper: Optional[_EntityBindKey[_O]] = None, clause: Optional[] = None, bind: Optional[_SessionBind] = None, **kw: Any) → Union[Engine, ]

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

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

Note

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

The pattern introduced at Custom Vertical Partitioning illustrates how to apply a custom bind-lookup scheme to a given a set of Engine objects. To apply a corresponding implementation for use with a AsyncSession and objects, continue to subclass Session and apply it to using AsyncSession.sync_session_class. The inner method must continue to return instances, which can be acquired from a AsyncEngine using the attribute:

```

using example from “Custom Vertical Partitioning”

import random
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.orm import Session
# construct async engines w/ async drivers
engines = {
    'leader':create_async_engine("sqlite+aiosqlite:///leader.db"),
    'other':create_async_engine("sqlite+aiosqlite:///other.db"),
    'follower1':create_async_engine("sqlite+aiosqlite:///follower1.db"),
    'follower2':create_async_engine("sqlite+aiosqlite:///follower2.db"),
}
class RoutingSession(Session):
    def get_bind(self, mapper=None, clause=None, **kw):
        # within get_bind(), return sync engines
        if mapper and issubclass(mapper.class_, MyOtherClass):
            return engines['other'].sync_engine
        elif self._flushing or isinstance(clause, (Update, Delete)):
            return engines['leader'].sync_engine
        else:
            return engines[
                random.choice(['follower1','follower2'])
            ].sync_engine
# apply to AsyncSession using sync_session_class
AsyncSessionMaker = async_sessionmaker(
    sync_session_class=RoutingSession
)
```
The [Session.get\_bind()]($694f628462946390.md#sqlalchemy.orm.Session.get_bind "sqlalchemy.orm.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()](#sqlalchemy.ext.asyncio.AsyncSession.run_sync "sqlalchemy.ext.asyncio.AsyncSession.run_sync"), so routines that wish to run SQL commands inside of [Session.get\_bind()]($694f628462946390.md#sqlalchemy.orm.Session.get_bind "sqlalchemy.orm.Session.get_bind") can continue to do so using blocking-style code, which will be translated to implicitly async calls at the point of invoking IO on the database drivers.

method get_nested_transaction() → Optional[AsyncSessionTransaction]

Return the current nested transaction in progress, if any.
- Returns:
  
  an object, or None.
New in version 1.4.18.
method sqlalchemy.ext.asyncio.AsyncSession.get_transaction() → Optional[]

Return the current root transaction in progress, if any.
- Returns:
  
  an AsyncSessionTransaction object, or None.
New in version 1.4.18.
classmethod identity_key(class\: Optional[Type[Any]] = None, _ident: Union[Any, Tuple[Any, …]] = None, *, instance: Optional[Any] = None, row: Optional[Union[[Any], RowMapping]] = None, identity_token: Optional[Any] = None) → _IdentityKeyType[Any]

Return an identity key.

Proxied for the class on behalf of the AsyncSession class.

This is an alias of .
attribute sqlalchemy.ext.asyncio.AsyncSession.identity_map

Proxy for the attribute on behalf of the AsyncSession class.
method in_nested_transaction() → bool

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

Proxied for the class on behalf of the AsyncSession class.

New in version 1.4.
method in_transaction() → bool

Return True if this Session has begun a transaction.

Proxied for the class on behalf of the AsyncSession class.

New in version 1.4.

See also
attribute sqlalchemy.ext.asyncio.AsyncSession.info

A user-modifiable dictionary.

Proxied for the class on behalf of the AsyncSession class.

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

Close this Session, using connection invalidation.

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

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

Proxied for the class on behalf of the AsyncSession class.

Changed in version 1.4: The 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 has failed, and that the Session.rollback() method must be emitted in order to fully roll back the transaction.

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

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

See also

“This Session’s transaction has been rolled back due to a previous exception during flush.” (or similar)
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 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 committed value, if any.

It is in effect a more expensive and accurate version of checking for the given instance in the 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 , but ultimately the state is the same as that loaded from the database, resulting in no net change here.
- Scalar attributes may not have recorded the previously set value when a new value was applied, if the attribute was not loaded, or was expired, at the time the new value was received - in these cases, the attribute is assumed to have a change, even if there is ultimately no net change against its database value. SQLAlchemy in most cases does not need the “old” value when a set event occurs, so it skips the expense of a SQL call if the old value isn’t present, based on the assumption that an UPDATE of the scalar value is usually needed, and in those few cases where it isn’t, is less expensive on average than issuing a defensive SELECT.
  
  The “old” value is fetched unconditionally upon set only if the attribute container has the active_history flag set to True. This flag is set typically for primary key attributes and scalar object references that are not a simple many-to-one. To set this flag for any arbitrary mapped column, use the active_history argument with column_property().
- Parameters:
  - instance – mapped instance to be tested for pending changes.
  - include_collections – Indicates if multivalued collections should be included in the operation. Setting this to False is a way to detect only local-column based properties (i.e. scalar columns or many-to-one foreign keys) that would result in an UPDATE for this instance upon flush.

method async merge(instance: _O, *, load: bool = True, options: Optional[Sequence[ORMOption]] = None) → _O

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

See also

Session.merge() - main documentation for merge
attribute new

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

Proxied for the Session class on behalf of the class.
attribute sqlalchemy.ext.asyncio.AsyncSession.no_autoflush

Return a context manager that disables autoflush.

Proxied for the 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 object_session(instance: object) → Optional[Session]

Return the to which an object belongs.

Proxied for the Session class on behalf of the class.

This is an alias of object_session().
method async refresh(instance: object, attribute_names: Optional[Iterable[str]] = None, with_for_update: Optional[ForUpdateArg] = None) → None

Expire and refresh the attributes on the given instance.

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

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

See also

- main documentation for refresh
method sqlalchemy.ext.asyncio.AsyncSession.async rollback() → None

Rollback the current transaction in progress.
method async run_sync(fn: Callable[[…], Any], *arg: Any, **kw: Any) → Any

Invoke the given sync callable passing sync self as the first argument.

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.

E.g.:

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.

See also

Running Synchronous Methods and Functions under asyncio
method async scalar(statement: Executable, params: Optional[_CoreSingleExecuteParams] = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: Optional[_BindArguments] = None, **kw: Any) → Any

Execute a statement and return a scalar result.

See also

- main documentation for scalar
method sqlalchemy.ext.asyncio.AsyncSession.async scalars(statement: , params: Optional[_CoreSingleExecuteParams] = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: Optional[_BindArguments] = None, **kw: Any) → ScalarResult[Any]

Execute a statement and return scalar results.
- Returns:
  
  a object
New in version 1.4.24.

See also

Session.scalars() - main documentation for scalars

- streaming version
method sqlalchemy.ext.asyncio.AsyncSession.async stream(statement: , params: Optional[_CoreAnyExecuteParams] = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: Optional[_BindArguments] = None, **kw: Any) → AsyncResult[Any]

Execute a statement and return a streaming object.
method sqlalchemy.ext.asyncio.AsyncSession.async stream_scalars(statement: , params: Optional[_CoreSingleExecuteParams] = None, *, execution_options: OrmExecuteOptionsParameter = {}, bind_arguments: Optional[_BindArguments] = None, **kw: Any) → AsyncScalarResult[Any]

Execute a statement and return a stream of scalar results.
- Returns:
  
  an object
New in version 1.4.24.

See also

Session.scalars() - main documentation for scalars

- non streaming version
attribute sqlalchemy.ext.asyncio.AsyncSession.sync_session:

Reference to the underlying Session this proxies requests towards.

This instance can be used as an event target.

See also

Using events with the asyncio extension

class sqlalchemy.ext.asyncio.AsyncSessionTransaction

A wrapper for the ORM object.

This object is provided so that a transaction-holding object for the AsyncSession.begin() may be returned.

The object supports both explicit calls to and AsyncSessionTransaction.rollback(), as well as use as an async context manager.

New in version 1.4.

Members

, rollback()

Class signature

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

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

Asynchronous I/O (asyncio)

Asynchronous I/O (asyncio)

Synopsis - Core

Synopsis - ORM

Using multiple asyncio event loops

Using asyncio scoped session

Engine API Documentation

Result Set API Documentation

using example from “Custom Vertical Partitioning”

using example from “Custom Vertical Partitioning”