Asynchronous I/O (asyncio)¶

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:

  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:

  # create AsyncSession with expire_on_commit=False
  async_session = AsyncSession(engine, expire_on_commit=False)
  
  # sessionmaker version
  async_session = sessionmaker(
      engine, expire_on_commit=False, class_=AsyncSession
  )
  
  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 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:

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

Other guidelines include:

• Methods like AsyncSession.expire() should be avoided in favor of AsyncSession.refresh()

• Avoid using the all cascade option documented at Cascades 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 AsyncSession.refresh() 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. A future release may introduce the ability to indicate eager loader options when invoking Session.refresh() and/or AsyncSession.refresh().

• Appropriate loader options should be employed for deferred() columns, if used at all, in addition to that of relationship() constructs as noted above. See Deferred Column Loading 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 AsyncSession.run_sync() 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()

  See also

  Making use of “dynamic” relationship loads without using Query - notes on migration to 2.0 style

Running Synchronous Methods and Functions under asyncio¶

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:

import asyncio

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


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

    """

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

    stmt = select(A)

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

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

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

    a1.data = "new data"


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

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

        await session.run_sync(fetch_and_update_objects)

        await session.commit()

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


asyncio.run(async_main())

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

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

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, ...):
    dbapi_connection.run_async(
        lambda connection: connection.set_type_codec(
            "MyCustomType", encoder, decoder, ...
        )
    )

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