Метод subquery
Метод subquery класса Query преобразует текущий
ORM-запрос в подзапрос, пригодный для использования в других
SQL-выражениях. Метод возвращает объект Alias, содержащий
подзапрос и его колонки. Параметров метод не принимает. Подзапрос
позволяет строить сложные запросы с вложенными выборками,
агрегациями и соединениями.
Синтаксис
query.subquery()
Пример
Давайте создадим модель и получим из запроса подзапрос, а затем выведем его SQL-представление:
from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
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]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
query = session.query(Article).filter(Article.status == 'published')
subq = query.subquery()
print(subq)
Результат выполнения кода:
"SELECT articles.id AS id, articles.title AS title, articles.text AS text, articles.status AS status, articles.num AS num
FROM articles
WHERE articles.status = ?) AS anon_1"
Пример
Давайте используем подзапрос для фильтрации внешнего запроса по количеству статей:
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
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]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add_all([
Article(title='article 1', text='text 1', status='published', num=1),
Article(title='article 2', text='text 2', status='draft', num=2),
Article(title='article 3', text='text 3', status='published', num=3),
])
session.commit()
subq = (
session.query(Article.status, func.count(Article.id).label('cnt'))
.group_by(Article.status)
.subquery()
)
res = (
session.query(subq.c.status, subq.c.cnt)
.filter(subq.c.cnt > 1)
.all()
)
print(res)
Результат выполнения кода:
[('published', 2)]
Пример
Давайте применим подзапрос в соединении с основной таблицей:
from sqlalchemy import create_engine, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
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]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add_all([
Article(title='article 1', text='text 1', status='published', num=1),
Article(title='article 2', text='text 2', status='draft', num=2),
])
session.commit()
subq = (
session.query(Article.status, func.max(Article.num).label('max_num'))
.group_by(Article.status)
.subquery()
)
res = (
session.query(Article.title)
.join(subq, Article.status == subq.c.status)
.filter(Article.num == subq.c.max_num)
.all()
)
print(res)
Результат выполнения кода:
[('article 1',), ('article 2',)]
