Атрибут sync_session

Атрибут sync_session класса AsyncSession предоставляет доступ к синхронному объекту Session, который обёрнут внутри асинхронной сессии. Асинхронная сессия в SQLAlchemy не является самостоятельной реализацией - она содержит внутри себя обычную синхронную сессию и делегирует ей большую часть работы. Атрибут sync_session возвращает именно эту внутреннюю сессию, что позволяет обращаться к синхронному API, когда асинхронный вариант метода отсутствует или неудобен. Атрибут доступен только для чтения и не принимает параметров.

Синтаксис

async_session.sync_session

Пример

Давайте создадим асинхронную сессию и получим из неё синхронную сессию через атрибут sync_session:

import asyncio

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

async def main():
	engine = create_async_engine('sqlite+aiosqlite:///:memory:')
	async with AsyncSession(engine) as session:
		print(session.sync_session)

asyncio.run(main())

Результат выполнения кода:

"<sqlalchemy.orm.session.Session object at 0x...>"

Пример

Атрибут sync_session позволяет проверить, к какому движку привязана сессия, через синхронный атрибут bind:

import asyncio

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

async def main():
	engine = create_async_engine('sqlite+aiosqlite:///:memory:')
	async with AsyncSession(engine) as session:
		print(session.sync_session.bind)

asyncio.run(main())

Результат выполнения кода:

"Engine(sqlite+aiosqlite:///:memory:)"

Пример

Синхронную сессию удобно использовать для доступа к методам, которые не имеют асинхронного аналога. Например, получим идентификатор текущей транзакции:

import asyncio

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

async def main():
	engine = create_async_engine('sqlite+aiosqlite:///:memory:')
	async with AsyncSession(engine) as session:
		await session.execute(text('SELECT 1'))
		print(session.sync_session.in_transaction())

asyncio.run(main())

Результат выполнения кода:

True

Пример

Также через sync_session можно работать с картой идентичности сессии и посмотреть количество загруженных объектов:

import asyncio

from sqlalchemy import Integer, String
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
	pass

class Article(Base):
	__tablename__ = 'articles'
	
	id: Mapped[int] = mapped_column(Integer, primary_key=True)
	title: Mapped[str] = mapped_column(String)

async def main():
	engine = create_async_engine('sqlite+aiosqlite:///:memory:')
	async with engine.begin() as conn:
		await conn.run_sync(Base.metadata.create_all)
	
	async with AsyncSession(engine) as session:
		article = Article(title='article')
		session.add(article)
		await session.flush()
		print(len(session.sync_session.identity_map))

asyncio.run(main())

Результат выполнения кода:

1

Смотрите также

класс AsyncSession,
который представляет асинхронную сессию
метод run_sync,
который выполняет синхронную функцию в асинхронном контексте
атрибут bind,
который возвращает движок сессии
атрибут identity_map,
который хранит загруженные объекты сессии