Асинхронный ввод/вывод (asyncio)¶

Предотвращение неявного ввода-вывода при использовании AsyncSession¶

Используя традиционный asyncio, приложение должно избегать точек, в которых может произойти обращение к IO-на-атрибутах. Для предотвращения этого принимаются следующие меры:

• Загрузчик нетерпения selectinload() используется для нетерпеливой загрузки коллекции A.bs в пределах области действия вызова await session.execute():

  stmt = select(A).options(selectinload(A.bs))

  Если бы стратегия загрузчика по умолчанию «lazyload» была оставлена на месте, доступ к атрибуту A.bs вызвал бы исключение asyncio. Существует множество вариантов ORM-загрузчика, которые могут быть настроены на уровне связки по умолчанию или использоваться на основе каждого запроса, что описано в Техники загрузки отношений.

• AsyncSession настроен с помощью Session.expire_on_commit, установленного в False, так что мы можем получить доступ к атрибутам объекта после вызова AsyncSession.commit(), как в строке в конце, где мы получаем доступ к атрибуту:

  # 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)

• Значение Column.server_default в столбце created_at по умолчанию не обновляется после INSERT; вместо этого оно обычно expired so that it can be loaded when needed. Аналогичное поведение применимо к столбцу, в котором параметр Column.default присвоен объекту выражения SQL. Чтобы получить доступ к этому значению с помощью asyncio, оно должно быть обновлено в процессе flush, что достигается установкой параметра mapper.eager_defaults на отображении:

  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}

Другие рекомендации включают:

• Следует избегать методов типа AsyncSession.expire() в пользу AsyncSession.refresh().

• Избегайте использования опции каскада all, задокументированной в Каскады, в пользу явного перечисления желаемых свойств каскада. Опция каскада all подразумевает, помимо прочего, настройку refresh-expire, что означает, что метод AsyncSession.refresh() уничтожит атрибуты на связанных объектах, но не обязательно обновит эти связанные объекты, если в relationship() не настроена ускоренная загрузка, оставив их в просроченном состоянии. В будущем выпуске может появиться возможность указывать опции ускоренной загрузки при вызове Session.refresh() и/или AsyncSession.refresh().

• Для столбцов deferred(), если они вообще используются, следует применять соответствующие опции загрузчика в дополнение к опциям для конструкций relationship(), как отмечено выше. См. раздел Отложенная загрузка колонн для получения информации об отложенной загрузке столбцов.

• Стратегия «динамического» загрузчика отношений, описанная в Динамические загрузчики отношений, по умолчанию не совместима с подходом asyncio. Она может быть использована напрямую только при вызове метода AsyncSession.run_sync(), описанного в Запуск синхронных методов и функций в asyncio, или при использовании его атрибута .statement для получения обычного select:

  user = await session.get(User, 42)
  addresses = (await session.scalars(user.addresses.statement)).all()
  stmt = user.addresses.statement.where(
      Address.email_address.startswith("patrick")
  )
  addresses_filter = (await session.scalars(stmt)).all()

  См.также

  Использование «динамической» загрузки отношений без использования Query - заметки о переходе на стиль 2.0

Запуск синхронных методов и функций в asyncio¶

В качестве альтернативного средства интеграции традиционной «ленивой загрузки» SQLAlchemy в цикл событий asyncio, предоставляется опциональный метод, известный как AsyncSession.run_sync(), который будет запускать любую функцию Python внутри гринлета, где традиционные концепции синхронного программирования будут переведены на использование await, когда они достигнут драйвера базы данных. Гипотетический подход здесь заключается в том, что асинхронно-ориентированное приложение может упаковать методы, связанные с базой данных, в функции, которые вызываются с помощью AsyncSession.run_sync().

Изменяя приведенный выше пример, если бы мы не использовали selectinload() для коллекции A.bs, мы могли бы выполнить обработку этих обращений к атрибутам в отдельной функции:

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())

Приведенный выше подход к запуску определенных функций в рамках «синхронизирующего» бегуна имеет некоторые параллели с приложением, которое запускает приложение SQLAlchemy поверх библиотеки программирования на основе событий, такой как gevent. Различия заключаются в следующем:

1. В отличие от использования gevent, мы можем продолжать использовать стандартный цикл событий Python asyncio или любой пользовательский цикл событий без необходимости интегрироваться в цикл событий gevent.

2. Никакого «обезьянничания» не происходит. В приведенном выше примере используется настоящий драйвер asyncio, а лежащий в основе SQLAlchemy пул соединений также использует встроенный в Python asyncio.Queue для пула соединений.

3. Программа может свободно переключаться между кодом async/await и содержащимися функциями, использующими код sync, практически без ущерба для производительности. Нет ни «исполнителя потока», ни каких-либо дополнительных ожиданий или синхронизации.

4. Базовые сетевые драйверы также используют чистые концепции Python asyncio, не используются сторонние сетевые библиотеки, как это делают gevent и eventlet.

Использование методов драйвера только с ожиданием в пуле соединений и других событиях¶

Как говорилось в предыдущем разделе, обработчики событий, такие как обработчики событий PoolEvents, получают соединение «DBAPI» в стиле синхронизации, которое представляет собой объект-обертку, предоставляемый диалектами asyncio SQLAlchemy для адаптации базового соединения «драйвера» asyncio в такое, которое может быть использовано внутренними компонентами SQLAlchemy. Особый случай возникает, когда пользовательская реализация такого обработчика событий должна использовать конечное «драйверное» соединение напрямую, используя только ожидаемые методы этого драйверного соединения. Одним из таких примеров является метод .set_type_codec(), предоставляемый драйвером asyncpg.

Для того, чтобы приспособить этот вариант использования, класс SQLAlchemy AdaptedConnection предоставляет метод AdaptedConnection.run_async(), который позволяет вызывать ожидаемую функцию в «синхронном» контексте обработчика событий или другого внутреннего компонента SQLAlchemy. Этот метод является прямым аналогом метода AsyncConnection.run_sync(), который позволяет методу в стиле sync работать в режиме async.

AdaptedConnection.run_async() должна быть передана функция, которая примет внутреннее соединение «драйвера» как единственный аргумент и вернет awaitable, который будет вызван методом AdaptedConnection.run_async(). Саму переданную функцию не нужно объявлять как async; совершенно нормально, если она будет Python lambda:, так как возвращаемое значение awaitable будет вызвано после возврата:

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

engine = create_async_engine(...)


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

Выше, объект, передаваемый в обработчик события register_custom_types, является экземпляром AdaptedConnection, который предоставляет DBAPI-подобный интерфейс к базовому асинхронному объекту соединения на уровне драйвера. Метод AdaptedConnection.run_async() затем предоставляет доступ к ожидающей среде, в которой можно действовать с базовым соединением на уровне драйвера.