CTE Architecture

Structuring Queries with CTEs

As your data analysis becomes more complex, your SQL queries can grow into tangled, hard-to-read blocks of code. Subqueries help, but they often lead to deep nesting that's difficult to debug. Common Table Expressions (CTEs) offer a cleaner way to structure your logic. They let you define temporary, named result sets that you can reference within a single query, almost like creating temporary views on the fly.

A Common Table Expression (CTE) is a feature supported by most modern SQL engines that lets you break your query into smaller, readable chunks using the WITH keyword.

In SQLAlchemy, you create a CTE by building a standard select() statement and then calling the .cte() method on it. This transforms your selection into a reusable component.

from sqlalchemy import select, func
from my_models import orders # Assuming an ORM model or Table object

# Define the logic for the CTE
# Let's find total sales per customer
customer_sales_stmt = (
    select(
        orders.c.customer_id,
        func.sum(orders.c.amount).label("total_sales")
    )
    .group_by(orders.c.customer_id)
)

# Create the CTE object
customer_sales_cte = customer_sales_stmt.cte("customer_sales")

The variable customer_sales_cte now represents a temporary table-like structure named customer_sales containing customer_id and total_sales. You can now use this in a subsequent query just like any other table.

Chaining Multiple CTEs

The real power of CTEs comes from chaining them together. You can define one CTE, then define a second CTE that queries the first one. This creates a logical, step-by-step data processing pipeline within a single query, making each part of the analysis distinct and testable.

Let's build on our sales example. First, we'll aggregate daily sales. Then, we'll use a second CTE to rank those days by sales volume.

# CTE 1: Aggregate sales by date
daily_sales_stmt = (
    select(
        orders.c.order_date,
        func.sum(orders.c.amount).label("total_daily_sales")
    )
    .group_by(orders.c.order_date)
)
daily_sales_cte = daily_sales_stmt.cte("daily_sales")

# CTE 2: Rank daily sales using the first CTE
sales_rank_stmt = (
    select(
        daily_sales_cte.c.order_date,
        daily_sales_cte.c.total_daily_sales,
        func.rank().over(order_by=daily_sales_cte.c.total_daily_sales.desc()).label("sales_rank")
    )
)
sales_rank_cte = sales_rank_stmt.cte("sales_with_rank")

# Final query using the second CTE
final_query = select(sales_rank_cte).where(sales_rank_cte.c.sales_rank <= 5)

Notice how sales_rank_stmt selects from daily_sales_cte. This modular approach makes the logic for calculating daily totals separate from the logic for ranking them. The final query is simple because all the heavy lifting was done in well-defined steps.

Working with the ORM and Aliases

CTEs integrate seamlessly with the SQLAlchemy ORM. You can join your ORM-mapped classes directly to a CTE. This is useful for enriching aggregated data with details from your main tables, like getting customer names for our top sales list.

However, joining CTEs with ORM models can lead to column name collisions. For example, your User model and a CTE might both have an id column. To resolve this, you must alias the ORM model.

from sqlalchemy.orm import aliased
from my_models import User # Assuming an ORM class

# Reuse our customer_sales_cte from the first example
# customer_sales_cte = ...

# Alias the ORM model to prevent name conflicts
UserAlias = aliased(User)

# Join the CTE with the aliased User model
query = (
    select(
        UserAlias.name,
        customer_sales_cte.c.total_sales
    )
    .join(UserAlias, UserAlias.id == customer_sales_cte.c.customer_id)
    .order_by(customer_sales_cte.c.total_sales.desc())
)

Using aliased(User) creates a proxy for the User class that can be safely used in a join. This ensures SQLAlchemy generates SQL with proper table aliases, preventing ambiguity in the SELECT and JOIN clauses.

Use Case Data Deduplication

A classic use for CTEs is data deduplication. The strategy involves using a window function like ROW_NUMBER() to identify duplicate records and then selecting only the unique ones. A CTE is perfect for staging this logic.

from sqlalchemy import select, func, text
from my_models import duplicates # A table with duplicate entries

# CTE to number rows partitioned by columns that define a duplicate
dedupe_stmt = select(
    duplicates.c.id,
    duplicates.c.email,
    duplicates.c.data,
    func.row_number().over(
        partition_by=[duplicates.c.email],
        order_by=duplicates.c.id.desc()
    ).label("row_num")
).cte("deduped_data")

# Select only the first occurrence of each email
final_query = select(
    dedupe_stmt.c.id,
    dedupe_stmt.c.email,
    dedupe_stmt.c.data
).where(dedupe_stmt.c.row_num == 1)

Here, the CTE assigns a unique, incrementing number to each record within groups of identical emails. The main query then simply filters for records where this number is 1, effectively keeping only the most recent entry (based on the descending id) for each email address.

This pattern cleanly separates the logic for identifying duplicates from the final selection, making it a robust and readable solution.

Quiz Questions 1/5

What is the primary benefit of using a Common Table Expression (CTE) in a complex SQL query?

Quiz Questions 2/5

In SQLAlchemy, which method is called on a select() statement to transform it into a Common Table Expression?

By breaking queries into logical, reusable CTEs, you can build and maintain complex analytical pipelines with greater clarity and confidence.

Advanced SQL Mastery with SQLAlchemy