Метод cte класса Select
Метод cte класса Select создает
общее табличное выражение (Common Table Expression)
на основе текущего запроса. Первым параметром
метод принимает имя CTE. Параметр recursive
включает рекурсивный режим. Созданный объект
можно использовать в других запросах как
обычную таблицу.
Синтаксис
Select.cte(name, [recursive])
Пример
Давайте создадим простое CTE на основе запроса
к таблице articles и выведем его
SQL-представление:
from sqlalchemy import create_engine, select
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]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
stmt = select(Article.id, Article.title).where(Article.status == 'published')
cte_obj = stmt.cte('published_articles')
print(cte_obj)
Результат выполнения кода:
"published_articles"
Пример
Давайте выполним CTE и получим строки из базы данных:
from sqlalchemy import create_engine, select, insert
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]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with engine.begin() as conn:
conn.execute(insert(Article), [
{'title': 'article 1', 'text': 'text 1', 'status': 'published', 'num': 1},
{'title': 'article 2', 'text': 'text 2', 'status': 'draft', 'num': 2},
{'title': 'article 3', 'text': 'text 3', 'status': 'published', 'num': 3},
])
stmt = select(Article.id, Article.title).where(Article.status == 'published')
cte_obj = stmt.cte('published_articles')
with engine.connect() as conn:
res = conn.execute(select(cte_obj)).fetchall()
print(res)
Результат выполнения кода:
[(1, 'article 1'), (3, 'article 3')]
Пример
Давайте используем CTE в соединении с основной таблицей:
from sqlalchemy import create_engine, select, insert
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]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with engine.begin() as conn:
conn.execute(insert(Article), [
{'title': 'article 1', 'text': 'text 1', 'status': 'published', 'num': 1},
{'title': 'article 2', 'text': 'text 2', 'status': 'draft', 'num': 2},
])
published = select(Article.id, Article.title).where(Article.status == 'published').cte('published_articles')
stmt = select(Article.title).join(published, Article.id == published.c.id)
with engine.connect() as conn:
res = conn.execute(stmt).fetchall()
print(res)
Результат выполнения кода:
[('article 1',)]
Пример
Давайте создадим рекурсивное CTE
с параметром recursive:
<+python+>
from sqlalchemy import create_engine, select, literal, column, Integer
engine = create_engine('sqlite:///:memory:')
with engine.begin() as conn:
conn.execute(literal("CREATE TABLE nums (n INTEGER)"))
conn.execute(literal("INSERT INTO nums (n) VALUES (1), (2), (3)"))
nums = select(column('n', Integer)).select_from(literal("nums")).cte('nums_cte', recursive=True)
print(nums)
<-python+>
Итоговый SQL запрос:
WITH RECURSIVE nums_cte AS (
SELECT n FROM nums
)
SELECT n FROM nums_cte
