Метод union
Метод union класса Query
объединяет результаты двух или более запросов в один
результирующий набор. Он соответствует SQL-оператору
UNION и удаляет дублирующиеся строки. Если нужно
сохранить дубликаты, используйте метод union_all.
Метод принимает один или несколько объектов Query
для объединения.
Синтаксис
query1.union(query2, query3, ...)
Пример
Давайте создадим два запроса и объединим их
с помощью метода union:
from sqlalchemy import create_engine
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]
status: Mapped[str]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add_all([
Article(title='article 1', status='published'),
Article(title='article 2', status='draft'),
Article(title='article 3', status='published'),
])
session.commit()
query1 = session.query(Article).filter(Article.status == 'published')
query2 = session.query(Article).filter(Article.title == 'article 2')
res = query1.union(query2).all()
for article in res:
print(article.title)
Результат выполнения кода:
"article 1"
"article 3"
"article 2"
Пример
Давайте объединим два запроса с одинаковой структурой и посмотрим на итоговый SQL-запрос:
from sqlalchemy import create_engine
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]
status: Mapped[str]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add_all([
Article(title='article 1', status='published'),
Article(title='article 2', status='draft'),
])
session.commit()
query1 = session.query(Article.title).filter(Article.status == 'published')
query2 = session.query(Article.title).filter(Article.status == 'draft')
res = query1.union(query2).all()
for row in res:
print(row.title)
Итоговый SQL запрос:
SELECT articles.title AS articles_title
FROM articles
WHERE articles.status = 'published'
UNION
SELECT articles.title AS articles_title
FROM articles
WHERE articles.status = 'draft'
