Метод cte класса Query
Метод cte класса Query
преобразует текущий запрос в общее табличное выражение
(Common Table Expression, CTE). CTE позволяет вынести
подзапрос в отдельную именованную конструкцию, к которой
можно обращаться в основном запросе. Метод принимает
необязательный параметр name - имя CTE,
а также параметр recursive, включающий рекурсивный режим.
Результатом работы метода является объект CTE,
который можно использовать в других запросах.
Синтаксис
Query.cte(name=None, recursive=False)
Пример
Давайте создадим простое CTE на основе запроса и выведем его 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]
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')
cte_obj = query.cte('published_articles')
print(cte_obj)
Результат выполнения кода:
"WITH published_articles AS (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 = ?) SELECT published_articles.id, published_articles.title, published_articles.text, published_articles.status, published_articles.num FROM published_articles"
Пример
Давайте используем CTE в основном запросе для выборки данных из него:
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]
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()
cte_obj = session.query(Article).filter(Article.status == 'published').cte('published_articles')
res = session.query(cte_obj.c.title).all()
print(res)
Результат выполнения кода:
[('article 1',), ('article 3',)]
Пример
Давайте создадим CTE с использованием параметра recursive
для рекурсивной выборки:
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]
text: Mapped[str]
status: Mapped[str]
num: Mapped[int]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
base_query = session.query(Article).filter(Article.num == 1)
cte_obj = base_query.cte('article_cte', recursive=True)
print(cte_obj)
Результат выполнения кода:
"WITH RECURSIVE article_cte AS (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.num = ?) SELECT article_cte.id, article_cte.title, article_cte.text, article_cte.status, article_cte.num FROM article_cte"
