Метод scalar класса AsyncSession

Метод scalar класса AsyncSession выполняет SQL-запрос и возвращает скалярное значение из первой строки результата. Первым параметром метод принимает объект запроса select или текстовый SQL-запрос. Если строк нет, возвращается значение None. Метод является асинхронным, поэтому вызывается через await.

Синтаксис

await session.scalar(statement, [params])

Пример

Давайте создадим таблицу articles и получим количество строк в ней:

import asyncio
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
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 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.scalar(
			select(func.count()).select_from(Article)
		)
		print(res)
	
	await engine.dispose()

asyncio.run(main())

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

3

Пример

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

import asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
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 AsyncSession(engine) as session:
		session.add(Article(title='article 1'))
		await session.commit()
		
		res = await session.scalar(
			select(Article.title).where(Article.id == 1)
		)
		print(res)
	
	await engine.dispose()

asyncio.run(main())

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

"article 1"

Пример

Если строк в результате нет, метод возвращает значение None:

import asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
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 AsyncSession(engine) as session:
		res = await session.scalar(
			select(Article.title).where(Article.id == 999)
		)
		print(res)
	
	await engine.dispose()

asyncio.run(main())

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

None

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

класс AsyncSession,
который представляет асинхронную сессию
метод execute,
который выполняет запрос и возвращает результат
метод scalars,
который возвращает поток скалярных значений
метод get,
который возвращает объект по первичному ключу