Метод get_one

Метод get_one класса AsyncSession выполняет запрос и возвращает ровно одну строку результата. Если запрос вернул ноль строк или больше одной строки, метод выбрасывает исключение NoResultFound или MultipleResultsFound соответственно. Первым параметром метод принимает SQL-выражение или ORM-запрос, а дополнительными параметрами можно передать значения для параметризации запроса. Метод является асинхронным, поэтому вызывается через await.

Синтаксис

await session.get_one(statement, params)

Пример

Давайте создадим асинхронный движок, модель Article и получим одну строку по идентификатору:

import asyncio
from sqlalchemy import Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy import select

class Base(DeclarativeBase):
	pass

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

async def main():
	engine = create_async_engine('sqlite:///: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',
			text='text 1',
			status='active',
			num=1
		))
		await session.commit()
		
		stmt = select(Article).where(Article.id == 1)
		article = await session.get_one(stmt)
		print(article.title)
	
	await engine.dispose()

asyncio.run(main())

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

"article 1"

Пример

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

import asyncio
from sqlalchemy import Integer, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

class Base(DeclarativeBase):
	pass

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

async def main():
	engine = create_async_engine('sqlite:///: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='active', num=1),
			Article(title='article 2', text='text 2', status='draft', num=2)
		])
		await session.commit()
		
		stmt = select(Article).where(Article.num == 2)
		article = await session.get_one(stmt)
		print(article.title)
	
	await engine.dispose()

asyncio.run(main())

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

"article 2"

Пример

Давайте посмотрим, что произойдет, если запрос вернет больше одной строки:

import asyncio
from sqlalchemy import Integer, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.exc import MultipleResultsFound

class Base(DeclarativeBase):
	pass

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

async def main():
	engine = create_async_engine('sqlite:///: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='active', num=1),
			Article(title='article 2', text='text 2', status='active', num=2)
		])
		await session.commit()
		
		stmt = select(Article).where(Article.status == 'active')
		try:
			article = await session.get_one(stmt)
			print(article.title)
		except MultipleResultsFound:
			print('multiple results found')
	
	await engine.dispose()

asyncio.run(main())

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

"multiple results found"

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

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