Класс AsyncSession
Класс AsyncSession представляет собой асинхронную
сессию для работы с базой данных. Он используется
в асинхронном коде с asyncio и позволяет
выполнять запросы, добавлять, изменять и удалять
объекты без блокировки event loop. Первым параметром
принимает объект AsyncEngine или фабрику
сессий async_sessionmaker. Сессия управляет
транзакциями, кэширует объекты в identity map и
предоставляет асинхронные версии методов обычной
сессии Session.
Синтаксис
AsyncSession(bind, [autoflush], [expire_on_commit])
Пример
Давайте создадим асинхронный движок и сессию
для базы SQLite в памяти:
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
async def main():
engine = create_async_engine('sqlite+aiosqlite:///:memory:')
async with AsyncSession(engine) as session:
print(session)
asyncio.run(main())
Результат выполнения кода:
"<sqlalchemy.ext.asyncio.session.AsyncSession object at 0x...>"
Пример
Давайте создадим модель и добавим запись через асинхронную сессию:
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]
text: Mapped[str]
status: Mapped[str]
num: Mapped[int]
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from models import Base, Article
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',
text='some text',
status='draft',
num=1
)
session.add(article)
await session.commit()
print(article.id)
asyncio.run(main())
Результат выполнения кода:
1
Пример
Давайте выполним асинхронный запрос к таблице
articles:
import asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from models import Base, Article
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', text='text 1', status='draft', num=1),
Article(title='article 2', text='text 2', status='published', num=2),
])
await session.commit()
res = await session.execute(select(Article))
for article in res.scalars():
print(article.title)
asyncio.run(main())
Результат выполнения кода:
['article 1', 'article 2']
Пример
Давайте откатим транзакцию при возникновении ошибки:
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from models import Base, Article
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:
try:
article = Article(
title='article',
text='some text',
status='draft',
num=1
)
session.add(article)
await session.flush()
raise Exception('something went wrong')
except Exception:
await session.rollback()
print('transaction rolled back')
asyncio.run(main())
Результат выполнения кода:
"transaction rolled back"
