Метод run_sync

Метод run_sync класса AsyncSession позволяет выполнить синхронную функцию, передав ей первым аргументом объект Session. Это удобно, когда нужно воспользоваться методами синхронной сессии или сторонними библиотеками, которые не поддерживают асинхронность. Первым параметром метод принимает вызываемую функцию, а остальные позиционные и именованные аргументы передаются в нее без изменений.

Синтаксис

await session.run_sync(fn, *args, **kwargs)

Пример

Давайте создадим асинхронную сессию и выполним синхронную функцию, которая возвращает список заголовков статей:

import asyncio
from sqlalchemy import select
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(primary_key=True)
	title: Mapped[str]

def get_titles(session):
	res = session.execute(select(Article.title)).scalars().all()
	return res

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:
		session.add_all([
			Article(title='article 1'),
			Article(title='article 2'),
			Article(title='article 3'),
		])
		await session.commit()
		
		res = await session.run_sync(get_titles)
		print(res)

asyncio.run(main())

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

['article 1', 'article 2', 'article 3']

Пример

Давайте передадим дополнительные аргументы в синхронную функцию через run_sync:

<+python+> import asyncio from sqlalchemy import select 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(primary_key=True) title: Mapped[str] def get_articles_by_status(session, status): stmt = select(Article.title).where(Article.status == status) res = session.execute(stmt).scalars().all() return res 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: session.add_all([ Article(title='article 1', status='published'), Article(title='article 2', status='draft'), Article(title='article 3', status='published'), ]) await session.commit() res = await session.run_sync( get_articles_by_status, 'published' ) print(res) asyncio.run(main()) <-python+>

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

['article 1', 'article 3']

Пример

Давайте используем run_sync для создания таблиц через метаданные модели:

<+python+> import asyncio from sqlalchemy.ext.asyncio import 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(primary_key=True) title: Mapped[str] 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 engine.connect() as conn: res = await conn.run_sync( lambda sync_conn: sync_conn.dialect.name ) print(res) asyncio.run(main()) <-python+>

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

"sqlite"

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

класс AsyncSession,
который представляет асинхронную сессию
метод execute,
который выполняет SQL-запрос асинхронно
метод commit,
который фиксирует транзакцию
атрибут sync_session,
который возвращает синхронную сессию