Функция union
Функция union объединяет результаты двух или более
запросов в один набор строк. Она принимает один или
несколько запросов в качестве аргументов и возвращает
объект CompoundSelect, который можно выполнить
как обычный запрос. В отличие от union_all
функция union удаляет дублирующиеся строки
из итогового результата.
Первый запрос передается первым аргументом, второй запрос вторым аргументом и так далее. Все объединяемые запросы должны возвращать одинаковое количество колонок с совместимыми типами данных.
Синтаксис
union(select1, select2, ...)
Пример
Давайте создадим таблицу и объединим результаты
двух запросов через union:
from sqlalchemy import create_engine, select, union
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()
stmt1 = select(Article.title).where(Article.status == 'published')
stmt2 = select(Article.title).where(Article.num > 1)
res = session.execute(union(stmt1, stmt2)).scalars().all()
print(res)
Результат выполнения кода:
['article 1', 'article 2', 'article 3']
Пример
Давайте убедимся, что union удаляет дубликаты,
а union_all их сохраняет:
from sqlalchemy import create_engine, select, union, union_all
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='published', num=2),
])
session.commit()
stmt1 = select(Article.title).where(Article.status == 'published')
stmt2 = select(Article.title).where(Article.num > 0)
res_union = session.execute(union(stmt1, stmt2)).scalars().all()
print(res_union)
res_union_all = session.execute(union_all(stmt1, stmt2)).scalars().all()
print(res_union_all)
Результат выполнения кода:
['article 1', 'article 2']
['article 1', 'article 2', 'article 1', 'article 2']
Пример
Давайте объединим три запроса и отсортируем итоговый результат:
from sqlalchemy import create_engine, select, union
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='archived', num=3),
])
session.commit()
stmt1 = select(Article.title).where(Article.status == 'published')
stmt2 = select(Article.title).where(Article.status == 'draft')
stmt3 = select(Article.title).where(Article.status == 'archived')
stmt = union(stmt1, stmt2, stmt3).order_by('title')
res = session.execute(stmt).scalars().all()
print(res)
Результат выполнения кода:
['article 1', 'article 2', 'article 3']
Пример
Давайте посмотрим итоговый SQL-запрос,
который генерирует union:
from sqlalchemy import create_engine, select, union
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]
status: Mapped[str]
num: Mapped[int]
stmt1 = select(Article.title).where(Article.status == 'published')
stmt2 = select(Article.title).where(Article.num > 1)
stmt = union(stmt1, stmt2)
print(stmt)
Итоговый SQL запрос:
SELECT articles.title
FROM articles
WHERE articles.status = :status_1
UNION
SELECT articles.title
FROM articles
WHERE articles.num > :num_1
