Функция async_scoped_session

Функция async_scoped_session создает объект сессии, привязанный к определенной области видимости (scope). Это позволяет автоматически переиспользовать одну и ту же сессию в рамках одной задачи или запроса. Первым параметром функция принимает фабрику сессий async_session_factory, вторым - функцию scopefunc, которая определяет текущую область видимости.

Синтаксис

async_scoped_session(async_session_factory, scopefunc)

Пример

Давайте создадим асинхронную сессию с областью видимости, используя asyncio.current_task в качестве функции области видимости:

import asyncio
from sqlalchemy.ext.asyncio import (
	create_async_engine,
	async_sessionmaker,
	async_scoped_session
)

engine = create_async_engine('sqlite+aiosqlite:///:memory:')
async_session_factory = async_sessionmaker(engine, expire_on_commit=False)
ScopedSession = async_scoped_session(
	async_session_factory,
	scopefunc=asyncio.current_task
)

async def main():
	session = ScopedSession()
	print(session)
	await ScopedSession.remove()
	await engine.dispose()

asyncio.run(main())

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

"<sqlalchemy.ext.asyncio.session.AsyncSession object at 0x...>"

Пример

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

import asyncio
from sqlalchemy import String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import (
	create_async_engine,
	async_sessionmaker,
	async_scoped_session
)

class Base(DeclarativeBase):
	pass

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

engine = create_async_engine('sqlite+aiosqlite:///:memory:')
async_session_factory = async_sessionmaker(engine, expire_on_commit=False)
ScopedSession = async_scoped_session(
	async_session_factory,
	scopefunc=asyncio.current_task
)

async def main():
	async with engine.begin() as conn:
		await conn.run_sync(Base.metadata.create_all)
	
	session = ScopedSession()
	article = Article(title='article 1', text='text 1')
	session.add(article)
	await session.commit()
	
	res = await session.execute(select(Article))
	articles = res.scalars().all()
	for a in articles:
		print(a.title)
	
	await ScopedSession.remove()
	await engine.dispose()

asyncio.run(main())

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

"article 1"

Пример

Давайте покажем, что сессия переиспользуется в рамках одной задачи. Два вызова ScopedSession вернут один и тот же объект:

import asyncio
from sqlalchemy.ext.asyncio import (
	create_async_engine,
	async_sessionmaker,
	async_scoped_session
)

engine = create_async_engine('sqlite+aiosqlite:///:memory:')
async_session_factory = async_sessionmaker(engine, expire_on_commit=False)
ScopedSession = async_scoped_session(
	async_session_factory,
	scopefunc=asyncio.current_task
)

async def main():
	session1 = ScopedSession()
	session2 = ScopedSession()
	print(session1 is session2)
	await ScopedSession.remove()
	await engine.dispose()

asyncio.run(main())

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

True

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

функцию create_async_engine,
которая создает асинхронный движок базы данных
функцию async_scoped_session,
которая управляет сессиями с областью видимости
класс AsyncAttrs,
который добавляет асинхронные атрибуты моделям