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

Метод scalars класса AsyncSession выполняет SQL-запрос и возвращает объект ScalarResult, который содержит значения первой колонки каждой строки результата. Первым параметром метод принимает SQL-выражение или ORM-запрос. Дополнительно можно передать параметры через именованный аргумент params. Метод является асинхронным, поэтому вызывается через await.

Метод удобен, когда из результата нужны только скалярные значения, например, список заголовков статей или идентификаторов, без лишней обертки в виде кортежей строк.

Синтаксис

await session.scalars(statement, [params])

Пример

Давайте создадим асинхронный движок, таблицу articles и получим список заголовков через метод scalars:

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]
	text: 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', text='text 1'),
			Article(title='article 2', text='text 2'),
			Article(title='article 3', text='text 3'),
		])
		await session.commit()
		
		res = await session.scalars(select(Article.title))
		print(res.all())

asyncio.run(main())

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

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

Пример

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

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]
	status: 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', status='published'),
			Article(title='article 2', status='draft'),
			Article(title='article 3', status='published'),
		])
		await session.commit()
		
		res = await session.scalars(
			select(Article.title).where(Article.status == 'published')
		)
		print(res.all())

asyncio.run(main())

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

['article 1', 'article 3']

Пример

Давайте переберем скалярные значения построчно с помощью асинхронного цикла:

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]

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'),
		])
		await session.commit()
		
		res = await session.scalars(select(Article.title))
		for title in res:
			print(title)

asyncio.run(main())

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

"article 1"
"article 2"

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

класс AsyncSession,
который представляет асинхронную сессию
метод execute,
который выполняет запрос и возвращает строки
метод scalar,
который возвращает одно скалярное значение
метод stream_scalars,
который отдает скалярные значения потоком