Asynchronous I/O (asyncio)

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

New in version 1.4.

Warning

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

See also

Asynchronous IO Support for Core and ORM - initial feature announcement

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

Asyncio Platform Installation Notes (Including Apple M1)

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

  1. x86_64 aarch64 ppc64le amd64 win32

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

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

  1. pip install sqlalchemy[asyncio]

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

Synopsis - Core

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

  1. import asyncio
  3. from sqlalchemy import Column
  4. from sqlalchemy import MetaData
  5. from sqlalchemy import select
  6. from sqlalchemy import String
  7. from sqlalchemy import Table
  8. from sqlalchemy.ext.asyncio import create_async_engine
  10. meta = MetaData()
  11. t1 = Table("t1", meta, Column("name", String(50), primary_key=True))
  14. async def async_main() -> None:
  15.     engine = create_async_engine(
  16.         "postgresql+asyncpg://scott:tiger@localhost/test",
  17.         echo=True,
  18.     )
  20.     async with engine.begin() as conn:
  21.         await conn.run_sync(meta.create_all)
  23.         await conn.execute(
  24.             t1.insert(), [{"name": "some name 1"}, {"name": "some name 2"}]
  25.         )
  27.     async with engine.connect() as conn:
  28.         # select a Result, which will be delivered with buffered
  29.         # results
  30.         result = await conn.execute(select(t1).where(t1.c.name == "some name 1"))
  32.         print(result.fetchall())
  34.     # for AsyncEngine created in function scope, close and
  35.     # clean-up pooled connections
  36.     await engine.dispose()
  39. asyncio.run(async_main())

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

Tip

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

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

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

Synopsis - ORM

Using 2.0 style querying, the AsyncSession class provides full ORM functionality. Within the default mode of use, special care must be taken to avoid lazy loading or other expired-attribute access involving ORM relationships and column attributes; the next section Preventing Implicit IO when Using AsyncSession details this. The example below illustrates a complete example including mapper and session configuration:

  1. from __future__ import annotations
  3. import asyncio
  4. import datetime
  6. from sqlalchemy import ForeignKey
  7. from sqlalchemy import func
  8. from sqlalchemy import select
  9. from sqlalchemy.ext.asyncio import async_sessionmaker
  10. from sqlalchemy.ext.asyncio import AsyncSession
  11. from sqlalchemy.ext.asyncio import create_async_engine
  12. from sqlalchemy.orm import DeclarativeBase
  13. from sqlalchemy.orm import Mapped
  14. from sqlalchemy.orm import mapped_column
  15. from sqlalchemy.orm import relationship
  16. from sqlalchemy.orm import selectinload
  19. class Base(DeclarativeBase):
  20.     pass
  23. class A(Base):
  24.     __tablename__ = "a"
  26.     id: Mapped[int] = mapped_column(primary_key=True)
  27.     data: Mapped[str]
  28.     create_date: Mapped[datetime.datetime] = mapped_column(server_default=func.now())
  29.     bs: Mapped[list[B]] = relationship()
  32. class B(Base):
  33.     __tablename__ = "b"
  34.     id: Mapped[int] = mapped_column(primary_key=True)
  35.     a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))
  36.     data: Mapped[str]
  39. async def insert_objects(async_session: async_sessionmaker[AsyncSession]) -> None:
  41.     async with async_session() as session:
  42.         async with session.begin():
  43.             session.add_all(
  44.                 [
  45.                     A(bs=[B(), B()], data="a1"),
  46.                     A(bs=[B()], data="a2"),
  47.                     A(bs=[B(), B()], data="a3"),
  48.                 ]
  49.             )
  52. async def select_and_update_objects(
  53.     async_session: async_sessionmaker[AsyncSession],
  54. ) -> None:
  56.     async with async_session() as session:
  57.         stmt = select(A).options(selectinload(A.bs))
  59.         result = await session.execute(stmt)
  61.         for a1 in result.scalars():
  62.             print(a1)
  63.             print(f"created at: {a1.create_date}")
  64.             for b1 in a1.bs:
  65.                 print(b1)
  67.         result = await session.execute(select(A).order_by(A.id).limit(1))
  69.         a1 = result.scalars().one()
  71.         a1.data = "new data"
  73.         await session.commit()
  75.         # access attribute subsequent to commit; this is what
  76.         # expire_on_commit=False allows
  77.         print(a1.data)
  80. async def async_main() -> None:
  81.     engine = create_async_engine(
  82.         "postgresql+asyncpg://scott:tiger@localhost/test",
  83.         echo=True,
  84.     )
  86.     # async_sessionmaker: a factory for new AsyncSession objects.
  87.     # expire_on_commit - don't expire objects after transaction commit
  88.     async_session = async_sessionmaker(engine, expire_on_commit=False)
  90.     async with engine.begin() as conn:
  91.         await conn.run_sync(Base.metadata.create_all)
  93.     await insert_objects(async_session)
  94.     await select_and_update_objects(async_session)
  96.     # for AsyncEngine created in function scope, close and
  97.     # clean-up pooled connections
  98.     await engine.dispose()
  101. asyncio.run(async_main())

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

Preventing Implicit IO when Using AsyncSession

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:

    1. 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 Relationship Loading Techniques.

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

    1. # create AsyncSession with expire_on_commit=False
    2. async_session = AsyncSession(engine, expire_on_commit=False)
    4. # sessionmaker version
    5. async_session = async_sessionmaker(engine, expire_on_commit=False)
    7. async with async_session() as session:
    8.     result = await session.execute(select(A).order_by(A.id))
    10.     a1 = result.scalars().first()
    12.     # commit would normally expire all attributes
    13.     await session.commit()
    15.     # access attribute subsequent to commit; this is what
    16.     # expire_on_commit=False allows
    17.     print(a1.data)

  • The Column.server_default 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 Column.default 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:

    1. class A(Base):
    2.     # ...
    4.     # column with a server_default, or SQL expression default
    5.     create_date = mapped_column(DateTime, server_default=func.now())
    7.     # add this so that it can be accessed
    8.     __mapper_args__ = {"eager_defaults": True}

Other guidelines include:

Running Synchronous Methods and Functions under asyncio

Deep Alchemy

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

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

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

  1. import asyncio
  3. from sqlalchemy import select
  4. from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
  7. def fetch_and_update_objects(session):
  8.     """run traditional sync-style ORM code in a function that will be
  9.     invoked within an awaitable.
  11.     """
  13.     # the session object here is a traditional ORM Session.
  14.     # all features are available here including legacy Query use.
  16.     stmt = select(A)
  18.     result = session.execute(stmt)
  19.     for a1 in result.scalars():
  20.         print(a1)
  22.         # lazy loads
  23.         for b1 in a1.bs:
  24.             print(b1)
  26.     # legacy Query use
  27.     a1 = session.query(A).order_by(A.id).first()
  29.     a1.data = "new data"
  32. async def async_main():
  33.     engine = create_async_engine(
  34.         "postgresql+asyncpg://scott:tiger@localhost/test",
  35.         echo=True,
  36.     )
  37.     async with engine.begin() as conn:
  38.         await conn.run_sync(Base.metadata.drop_all)
  39.         await conn.run_sync(Base.metadata.create_all)
  41.     async with AsyncSession(engine) as session:
  42.         async with session.begin():
  43.             session.add_all(
  44.                 [
  45.                     A(bs=[B(), B()], data="a1"),
  46.                     A(bs=[B()], data="a2"),
  47.                     A(bs=[B(), B()], data="a3"),
  48.                 ]
  49.             )
  51.         await session.run_sync(fetch_and_update_objects)
  53.         await session.commit()
  55.     # for AsyncEngine created in function scope, close and
  56.     # clean-up pooled connections
  57.     await engine.dispose()
  60. asyncio.run(async_main())

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

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

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

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

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

Using events with the asyncio extension

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

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

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

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

Examples of Event Listeners with Async Engines / Sessions / Sessionmakers

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

  • Core Events on AsyncEngine

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

    ``` import asyncio

    from sqlalchemy import event from sqlalchemy import text from sqlalchemy.engine import Engine from sqlalchemy.ext.asyncio import create_async_engine

    engine = create_async_engine(“postgresql+asyncpg://scott:tiger@localhost:5432/test”)

  1. # connect event on instance of Engine
  2. @event.listens_for(engine.sync_engine, "connect")
  3. def my_on_connect(dbapi_con, connection_record):
  4.     print("New DBAPI connection:", dbapi_con)
  5.     cursor = dbapi_con.cursor()
  7.     # sync style API use for adapted DBAPI connection / cursor
  8.     cursor.execute("select 'execute from event'")
  9.     print(cursor.fetchone()[0])
  12. # before_execute event on all Engine instances
  13. @event.listens_for(Engine, "before_execute")
  14. def my_before_execute(
  15.     conn,
  16.     clauseelement,
  17.     multiparams,
  18.     params,
  19.     execution_options,
  20. ):
  21.     print("before execute!")
  24. async def go():
  25.     async with engine.connect() as conn:
  26.         await conn.execute(text("select 1"))
  27.     await engine.dispose()
  30. asyncio.run(go())
  31. ```
  33. Output:
  35. ```
  36. New DBAPI connection: <AdaptedConnection <asyncpg.connection.Connection object at 0x7f33f9b16960>>
  37. execute from event
  38. before execute!
  39. ```

  • ORM Events on AsyncSession

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

    ``` import asyncio

    from sqlalchemy import event from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.orm import Session

    engine = create_async_engine(“postgresql+asyncpg://scott:tiger@localhost:5432/test”)

    session = AsyncSession(engine)

  1. # before_commit event on instance of Session
  2. @event.listens_for(session.sync_session, "before_commit")
  3. def my_before_commit(session):
  4.     print("before commit!")
  6.     # sync style API use on Session
  7.     connection = session.connection()
  9.     # sync style API use on Connection
  10.     result = connection.execute(text("select 'execute from event'"))
  11.     print(result.first())
  14. # after_commit event on all Session instances
  15. @event.listens_for(Session, "after_commit")
  16. def my_after_commit(session):
  17.     print("after commit!")
  20. async def go():
  21.     await session.execute(text("select 1"))
  22.     await session.commit()
  24.     await session.close()
  25.     await engine.dispose()
  28. asyncio.run(go())
  29. ```
  31. Output:
  33. ```
  34. before commit!
  35. execute from event
  36. after commit!
  37. ```

  • ORM Events on async_sessionmaker

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

    ``` import asyncio

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

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

  1. @event.listens_for(sync_maker, "before_commit")
  2. def before_commit(session):
  3.     print("before commit")
  6. async def main():
  7.     async_session = maker()
  9.     await async_session.commit()
  12. asyncio.run(main())
  13. ```
  15. Output:
  17. ```
  18. before commit
  19. ```

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:

  1.  SQLAlchemy    SQLAlchemy        SQLAlchemy          SQLAlchemy   plain
  2.   asyncio      asyncio           ORM/Core            asyncio      asyncio
  3.   (public      (internal)                            (internal)
  4.   facing)
  5. -------------|------------|------------------------|-----------|------------
  6. asyncio API  |            |                        |           |
  7. call  ->     |            |                        |           |
  8.              |  ->  ->    |                        |  ->  ->   |
  9.              |~~~~~~~~~~~~| sync API call ->       |~~~~~~~~~~~|
  10.              | asyncio    |  event hooks ->        | sync      |
  11.              | to         |   invoke action ->     | to        |
  12.              | sync       |    event hooks ->      | asyncio   |
  13.              | (greenlet) |     dialect ->         | (leave    |
  14.              |~~~~~~~~~~~~|      event hooks ->    | greenlet) |
  15.              |  ->  ->    |       sync adapted     |~~~~~~~~~~~|
  16.              |            |               DBAPI -> |  ->  ->   | asyncio
  17.              |            |                        |           | 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.

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

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

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

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

  1. from sqlalchemy import event
  2. from sqlalchemy.ext.asyncio import create_async_engine
  4. engine = create_async_engine(...)
  7. @event.listens_for(engine.sync_engine, "connect")
  8. def register_custom_types(dbapi_connection, *args):
  9.     dbapi_connection.run_async(
  10.         lambda connection: connection.set_type_codec(
  11.             "MyCustomType",
  12.             encoder,
  13.             decoder,  # ...
  14.         )
  15.     )

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

New in version 1.4.30.

Using multiple asyncio event loops

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

If an AsyncEngine is be passed from one event loop to another, the method AsyncEngine.dispose() should be called before it’s re-used on a new event loop. Failing to do so may lead to a 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:

  1. from sqlalchemy.ext.asyncio import create_async_engine
  2. from sqlalchemy.pool import NullPool
  4. engine = create_async_engine(
  5.     "postgresql+asyncpg://user:pass@host/dbname",
  6.     poolclass=NullPool,
  7. )

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

Tip

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

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

  1. from asyncio import current_task
  3. from sqlalchemy.ext.asyncio import (
  4.     async_scoped_session,
  5.     async_sessionmaker,
  6. )
  8. async_session_factory = async_sessionmaker(
  9.     some_async_engine,
  10.     expire_on_commit=False,
  11. )
  12. AsyncScopedSession = async_scoped_session(
  13.     async_session_factory,
  14.     scopefunc=current_task,
  15. )
  16. 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 AsyncSession is accessed. The function should therefore be idempotent and lightweight, and should not attempt to create or mutate any state, such as establishing callbacks, etc.

Warning

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

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

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

New in version 1.4.19.

Using the Inspector to inspect schema objects

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

  1. import asyncio
  3. from sqlalchemy import inspect
  4. from sqlalchemy.ext.asyncio import create_async_engine
  6. engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost/test")
  9. def use_inspector(conn):
  10.     inspector = inspect(conn)
  11.     # use the inspector
  12.     print(inspector.get_view_names())
  13.     # return any value to the caller
  14.     return inspector.get_table_names()
  17. async def async_main():
  18.     async with engine.connect() as conn:
  19.         tables = await conn.run_sync(use_inspector)
  22. asyncio.run(async_main())

See also

Reflecting Database Objects

Runtime Inspection API

Engine API Documentation

Object NameDescription

async_engine_from_config(configuration[, prefix], kwargs)

Create a new AsyncEngine instance using a configuration dictionary.

AsyncConnection

An asyncio proxy for a Connection.

AsyncEngine

An asyncio proxy for a Engine.

AsyncTransaction

An asyncio proxy for a Transaction.

create_async_engine(url, kw)

Create a new async engine instance.

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

Create a new async engine instance.

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

New in version 1.4.

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

Create a new AsyncEngine instance using a configuration dictionary.

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

New in version 1.4.29.

class sqlalchemy.ext.asyncio.AsyncEngine

An asyncio proxy for a Engine.

AsyncEngine is acquired using the create_async_engine() function:

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

New in version 1.4.

Members

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

Class signature

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

class sqlalchemy.ext.asyncio.AsyncConnection

An asyncio proxy for a Connection.

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

  1. from sqlalchemy.ext.asyncio import create_async_engine
  2. engine = create_async_engine("postgresql+asyncpg://user:pass@host/dbname")
  4. async with engine.connect() as conn:
  5.     result = await conn.execute(select(table))

New in version 1.4.

Members

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

Class signature

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

class sqlalchemy.ext.asyncio.AsyncTransaction

An asyncio proxy for a Transaction.

Members

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

Class signature

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

Result Set API Documentation

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

Object NameDescription

AsyncMappingResult

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

AsyncResult

An asyncio wrapper around a Result object.

AsyncScalarResult

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

AsyncTupleResult

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

class sqlalchemy.ext.asyncio.AsyncResult

An asyncio wrapper around a Result object.

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

Note

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

New in version 1.4.

Members

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

Class signature

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

class sqlalchemy.ext.asyncio.AsyncScalarResult

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

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

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

New in version 1.4.

Members

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

Class signature

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

class sqlalchemy.ext.asyncio.AsyncMappingResult

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

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

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

New in version 1.4.

Members

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

Class signature

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

class sqlalchemy.ext.asyncio.AsyncTupleResult

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

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

Class signature

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

ORM Session API Documentation

Object NameDescription

async_object_session(instance)

Return the AsyncSession to which the given instance belongs.

async_scoped_session

Provides scoped management of AsyncSession objects.

async_session(session)

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

async_sessionmaker

A configurable AsyncSession factory.

AsyncSession

Asyncio version of Session.

AsyncSessionTransaction

A wrapper for the ORM SessionTransaction object.

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

Return the AsyncSession to which the given instance belongs.

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

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

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

  • Parameters:

    instance – an ORM mapped instance

    Returns:

    an AsyncSession object, or None.

New in version 1.4.18.

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

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

New in version 1.4.18.

class sqlalchemy.ext.asyncio.async_sessionmaker

A configurable AsyncSession factory.

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

e.g.:

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

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

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

See also

Synopsis - ORM - shows example use

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

Members

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

Class signature

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

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

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

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

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

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

    Construct a new async_sessionmaker.

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

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

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

    e.g.:

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

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

    (Re)configure the arguments for this async_sessionmaker.

    e.g.:

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

class sqlalchemy.ext.asyncio.async_scoped_session

Provides scoped management of AsyncSession objects.

See the section Using asyncio scoped session for usage details.

New in version 1.4.19.

Members

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

Class signature

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

  1. See also
  3. [Refreshing / Expiring]($592600365cded3ff.md#session-expire) - introductory material
  5. [Session.expire()]($694f628462946390.md#sqlalchemy.orm.Session.expire "sqlalchemy.orm.Session.expire")
  7. [Session.refresh()]($694f628462946390.md#sqlalchemy.orm.Session.refresh "sqlalchemy.orm.Session.refresh")
  9. [Query.populate\_existing()]($3d0cc000ec6c7150.md#sqlalchemy.orm.Query.populate_existing "sqlalchemy.orm.Query.populate_existing")
  1. import random
  3. from sqlalchemy.ext.asyncio import AsyncSession
  4. from sqlalchemy.ext.asyncio import create_async_engine
  5. from sqlalchemy.ext.asyncio import async_sessionmaker
  6. from sqlalchemy.orm import Session
  8. # construct async engines w/ async drivers
  9. engines = {
  10.     'leader':create_async_engine("sqlite+aiosqlite:///leader.db"),
  11.     'other':create_async_engine("sqlite+aiosqlite:///other.db"),
  12.     'follower1':create_async_engine("sqlite+aiosqlite:///follower1.db"),
  13.     'follower2':create_async_engine("sqlite+aiosqlite:///follower2.db"),
  14. }
  16. class RoutingSession(Session):
  17.     def get_bind(self, mapper=None, clause=None, **kw):
  18.         # within get_bind(), return sync engines
  19.         if mapper and issubclass(mapper.class_, MyOtherClass):
  20.             return engines['other'].sync_engine
  21.         elif self._flushing or isinstance(clause, (Update, Delete)):
  22.             return engines['leader'].sync_engine
  23.         else:
  24.             return engines[
  25.                 random.choice(['follower1','follower2'])
  26.             ].sync_engine
  28. # apply to AsyncSession using sync_session_class
  29. AsyncSessionMaker = async_sessionmaker(
  30.     sync_session_class=RoutingSession
  31. )
  32. ```
  34. 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 sqlalchemy.ext.asyncio.async_scoped_session.identity_key(class\: Optional[Type[Any]] = None, _ident: Union[Any, Tuple[Any, …]] = None, *, instance: Optional[Any] = None, row: Optional[Union[Row[Any], RowMapping]] = None, identity_token: Optional[Any] = None) → _IdentityKeyType[Any]

    Return an identity key.

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

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

    This is an alias of identity_key().

  • attribute sqlalchemy.ext.asyncio.async_scoped_session.identity_map

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

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

  • attribute sqlalchemy.ext.asyncio.async_scoped_session.info

    A user-modifiable dictionary.

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

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

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

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

    Close this Session, using connection invalidation.

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

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

  • attribute sqlalchemy.ext.asyncio.async_scoped_session.is_active

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

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

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

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

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

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

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

    See also

    “This Session’s transaction has been rolled back due to a previous exception during flush.” (or similar)

    Session.in_transaction()

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

    Return True if the given instance has locally modified attributes.

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

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

    This method retrieves the history for each instrumented attribute on the instance and performs a comparison of the current value to its previously 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.:

    1. return session.is_modified(someobject)

    A few caveats to this method apply:

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

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

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

    • Parameters:

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

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

class sqlalchemy.ext.asyncio.AsyncSession

Asyncio version of Session.

The AsyncSession is a proxy for a traditional Session instance.

New in version 1.4.

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

Members

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

Class signature

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

  1. See also
  3. [Refreshing / Expiring]($592600365cded3ff.md#session-expire) - introductory material
  5. [Session.expire()]($694f628462946390.md#sqlalchemy.orm.Session.expire "sqlalchemy.orm.Session.expire")
  7. [Session.refresh()]($694f628462946390.md#sqlalchemy.orm.Session.refresh "sqlalchemy.orm.Session.refresh")
  9. [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 Session class on behalf of the AsyncSession class.

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

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

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

    See also

    Refreshing / Expiring - introductory material

    Session.expire()

    Session.refresh()

    Query.populate_existing()

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

    Remove the instance from this Session.

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

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

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

    Remove all object instances from this Session.

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

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

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

    Flush all the object changes to the database.

    See also

    Session.flush() - main documentation for flush

  • method sqlalchemy.ext.asyncio.AsyncSession.async get(entity: _EntityBindKey[_O], ident: _PKIdentityArgument, *, options: 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

    Session.get() - main documentation for get

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

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

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

    Note

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

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

    ```

    using example from “Custom Vertical Partitioning”

  1. import random
  3. from sqlalchemy.ext.asyncio import AsyncSession
  4. from sqlalchemy.ext.asyncio import create_async_engine
  5. from sqlalchemy.ext.asyncio import async_sessionmaker
  6. from sqlalchemy.orm import Session
  8. # construct async engines w/ async drivers
  9. engines = {
  10.     'leader':create_async_engine("sqlite+aiosqlite:///leader.db"),
  11.     'other':create_async_engine("sqlite+aiosqlite:///other.db"),
  12.     'follower1':create_async_engine("sqlite+aiosqlite:///follower1.db"),
  13.     'follower2':create_async_engine("sqlite+aiosqlite:///follower2.db"),
  14. }
  16. class RoutingSession(Session):
  17.     def get_bind(self, mapper=None, clause=None, **kw):
  18.         # within get_bind(), return sync engines
  19.         if mapper and issubclass(mapper.class_, MyOtherClass):
  20.             return engines['other'].sync_engine
  21.         elif self._flushing or isinstance(clause, (Update, Delete)):
  22.             return engines['leader'].sync_engine
  23.         else:
  24.             return engines[
  25.                 random.choice(['follower1','follower2'])
  26.             ].sync_engine
  28. # apply to AsyncSession using sync_session_class
  29. AsyncSessionMaker = async_sessionmaker(
  30.     sync_session_class=RoutingSession
  31. )
  32. ```
  34. 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 sqlalchemy.ext.asyncio.AsyncSession.get_nested_transaction() → Optional[AsyncSessionTransaction]

    Return the current nested transaction in progress, if any.

    New in version 1.4.18.

  • method sqlalchemy.ext.asyncio.AsyncSession.get_transaction() → Optional[AsyncSessionTransaction]

    Return the current root transaction in progress, if any.

    New in version 1.4.18.

  • classmethod sqlalchemy.ext.asyncio.AsyncSession.identity_key(class\: Optional[Type[Any]] = None, _ident: Union[Any, Tuple[Any, …]] = None, *, instance: Optional[Any] = None, row: Optional[Union[Row[Any], RowMapping]] = None, identity_token: Optional[Any] = None) → _IdentityKeyType[Any]

    Return an identity key.

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

    This is an alias of identity_key().

  • attribute sqlalchemy.ext.asyncio.AsyncSession.identity_map

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

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

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

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

    New in version 1.4.

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

    Return True if this Session has begun a transaction.

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

    New in version 1.4.

    See also

    Session.is_active

  • attribute sqlalchemy.ext.asyncio.AsyncSession.info

    A user-modifiable dictionary.

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

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

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

    Close this Session, using connection invalidation.

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

  • attribute sqlalchemy.ext.asyncio.AsyncSession.is_active

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

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

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

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

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

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

    See also

    “This Session’s transaction has been rolled back due to a previous exception during flush.” (or similar)

    Session.in_transaction()

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

    Return True if the given instance has locally modified attributes.

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

    This method retrieves the history for each instrumented attribute on the instance and performs a comparison of the current value to its previously 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.:

    1. return session.is_modified(someobject)

    A few caveats to this method apply:

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

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

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

    • Parameters:

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

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

class sqlalchemy.ext.asyncio.AsyncSessionTransaction

A wrapper for the ORM SessionTransaction object.

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

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

New in version 1.4.

Members

commit(), rollback()

Class signature

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