Метод query класса scoped_session

Метод query класса scoped_session создает объект Query, привязанный к текущей сессии текущего потока. Метод принимает ORM-класс или несколько ORM-классов и возвращает объект запроса, через который можно фильтровать, сортировать и извлекать данные. Если сессия для текущего потока еще не создана, scoped_session создаст ее автоматически.

Синтаксис

scoped_session.query(*entities)

Пример

Давайте создадим модель Article и получим все записи через метод query:

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, scoped_session, sessionmaker

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(100))

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

Session = scoped_session(sessionmaker(bind=engine))

Session.add(Article(title='article 1', text='text 1'))
Session.add(Article(title='article 2', text='text 2'))
Session.commit()

res = Session.query(Article).all()
for article in res:
	print(article.title)

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

"article 1"
"article 2"

Пример

Давайте применим фильтрацию через метод filter_by:

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, scoped_session, sessionmaker

class Base(DeclarativeBase):
	pass

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

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

Session = scoped_session(sessionmaker(bind=engine))

Session.add(Article(title='article 1', status='active'))
Session.add(Article(title='article 2', status='draft'))
Session.commit()

res = Session.query(Article).filter_by(status='active').all()
for article in res:
	print(article.title)

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

"article 1"

Пример

Давайте извлечем только одно поле через метод query:

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, scoped_session, sessionmaker

class Base(DeclarativeBase):
	pass

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

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

Session = scoped_session(sessionmaker(bind=engine))

Session.add(Article(title='article 1', num=10))
Session.add(Article(title='article 2', num=20))
Session.commit()

res = Session.query(Article.title).all()
for row in res:
	print(row.title)

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

"article 1"
"article 2"

Пример

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

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, scoped_session, sessionmaker

class Base(DeclarativeBase):
	pass

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

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

Session = scoped_session(sessionmaker(bind=engine))

Session.add(Article(title='article 1', num=30))
Session.add(Article(title='article 2', num=10))
Session.add(Article(title='article 3', num=20))
Session.commit()

res = Session.query(Article).order_by(Article.num).all()
for article in res:
	print(article.title, article.num)

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

"article 2 10"
"article 3 20"
"article 1 30"

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

класс scoped_session,
который управляет сессиями в разных потоках
метод remove,
который удаляет текущую сессию
метод configure,
который настраивает фабрику сессий
метод query,
который создает объект запроса для текущей сессии