Queries con select(): filter, join, group_by

Descripción

select() es el constructor universal de queries en SQLAlchemy 2.0. Reemplaza al session.query() de 1.x. Cualquier query SQL que escribiste en módulos 2-3 se puede expresar con select() — JOINs, subqueries, aggregations, window functions, todo.

Esta cápsula te enseña a construir queries complejas con la API moderna. Al final, podrás traducir cualquier query del módulo 2 a SQLAlchemy.

Sintaxis básica

from sqlalchemy import select
from app.models import User

stmt = select(User)
# SELECT users.id, users.email, ... FROM users

result = session.execute(stmt)
users = result.scalars().all()

select(User) retorna objetos User. Si quieres tuplas (columnas específicas):

stmt = select(User.username, User.email)
# SELECT users.username, users.email FROM users

result = session.execute(stmt)
rows = result.all()
# [Row(username='maria', email='maria@blog.local'), ...]

WHERE: filtrado

stmt = select(User).where(User.is_active == True)
# WHERE users.is_active = TRUE

# Múltiples condiciones (AND implícito)
stmt = select(User).where(
    User.is_active == True,
    User.email_verified == True,
)
# WHERE users.is_active = TRUE AND users.email_verified = TRUE

# AND/OR explícitos
from sqlalchemy import and_, or_

stmt = select(User).where(
    or_(
        User.is_admin == True,
        and_(User.is_active == True, User.email_verified == True),
    )
)

Operadores

⚠️ Nunca uses User.deleted_at == None — Python interpreta esto como un valor especial. Siempre User.deleted_at.is_(None).

ORDER BY, LIMIT, OFFSET

stmt = (
    select(Post)
    .where(Post.published == True)
    .order_by(Post.published_at.desc())
    .limit(10)
    .offset(20)  # paginación
)

.desc() y .asc() para dirección (default: ASC).

Múltiples columnas:

.order_by(Post.published.desc(), Post.published_at.desc())

JOINs

INNER JOIN

stmt = (
    select(Post)
    .join(User, Post.author_id == User.id)
    .where(User.username == "maria")
)
# SELECT posts.* FROM posts INNER JOIN users ON posts.author_id = users.id WHERE users.username = 'maria'

Si tienes relationship() configurada, puedes simplificar:

stmt = (
    select(Post)
    .join(Post.author)  # usa la relación
    .where(User.username == "maria")
)

LEFT JOIN (outer)

stmt = (
    select(User, Post)
    .outerjoin(Post, Post.author_id == User.id)
)
# LEFT OUTER JOIN

O con relación:

stmt = select(User).outerjoin(User.posts)

Múltiples joins

stmt = (
    select(Post.title, User.username, Category.name)
    .join(Post.author)
    .outerjoin(Post.category)
    .where(Post.published == True)
    .order_by(Post.published_at.desc())
)

SELECT con múltiples entidades / columnas

# Tuplas de objetos
stmt = select(Post, User).join(Post.author)
result = session.execute(stmt).all()
# [(Post, User), (Post, User), ...]

# Mezcla columnas y entidades
stmt = select(Post, User.username).join(Post.author)
# [(Post, "maria"), ...]

# Solo columnas
stmt = select(Post.title, Post.published_at)
# [Row(title=..., published_at=...), ...]

Para extraer:

result = session.execute(stmt)

# Si es UN modelo: scalars()
posts = result.scalars().all()

# Si son MÚLTIPLES cosas: iterar Rows
for post, username in session.execute(stmt):
    print(f"{username}: {post.title}")

Aggregations: COUNT, SUM, AVG, GROUP BY

from sqlalchemy import func

# Total de posts publicados
total = session.execute(
    select(func.count(Post.id)).where(Post.published == True)
).scalar_one()

# Promedio de longitud de títulos
avg_len = session.execute(
    select(func.avg(func.length(Post.title)))
).scalar_one()

# Posts por autor
stmt = (
    select(User.username, func.count(Post.id).label("num_posts"))
    .outerjoin(User.posts)
    .group_by(User.id, User.username)
    .order_by(func.count(Post.id).desc())
)

for username, num in session.execute(stmt):
    print(f"{username}: {num} posts")

func.X() accede a cualquier función SQL: func.count(), func.avg(), func.now(), func.lower(), func.coalesce(), func.string_agg(), etc.

HAVING

stmt = (
    select(User.username, func.count(Post.id).label("n"))
    .join(User.posts)
    .group_by(User.id, User.username)
    .having(func.count(Post.id) >= 2)
)
# HAVING count(...) >= 2

FILTER (PostgreSQL-specific)

stmt = (
    select(
        User.username,
        func.count(Post.id).label("total"),
        func.count(Post.id).filter(Post.published == True).label("publicados"),
        func.count(Post.id).filter(Post.published == False).label("drafts"),
    )
    .outerjoin(User.posts)
    .group_by(User.id, User.username)
)

func.count(...).filter(condition) traduce a count(...) FILTER (WHERE ...).

Subqueries

from sqlalchemy import select

# Subquery escalar
posts_count_per_user = (
    select(func.count(Post.id))
    .where(Post.author_id == User.id)
    .scalar_subquery()
)

stmt = select(User.username, posts_count_per_user.label("num_posts"))

IN con subquery

admin_ids = select(User.id).where(User.is_admin == True)

stmt = select(Post).where(Post.author_id.in_(admin_ids))

EXISTS

from sqlalchemy import exists

stmt = select(User).where(
    exists().where(Post.author_id == User.id, Post.published == True)
)

CTEs (WITH ... AS)

# CTE
post_counts = (
    select(Post.author_id, func.count().label("n"))
    .group_by(Post.author_id)
    .cte("post_counts")
)

# Query principal usando la CTE
stmt = (
    select(User.username, post_counts.c.n)
    .join(post_counts, User.id == post_counts.c.author_id)
)

post_counts.c.n accede a la columna n de la CTE (el c es como Table.c.column).

CTE recursiva

# Comments tree para un post específico
top_comments = (
    select(Comment)
    .where(Comment.post_id == post_id, Comment.parent_comment_id.is_(None))
    .cte("comment_tree", recursive=True)
)

# El "anchor" + recursive part
top_comments = top_comments.union_all(
    select(Comment).join(top_comments, Comment.parent_comment_id == top_comments.c.id)
)

stmt = select(top_comments)

CTEs recursivas son raras pero potentes. Para el blog, una librería como sqlalchemy-utils o queries SQL directos pueden ser más legibles.

Funciones útiles del blog

Top N por categoría (window function)

from sqlalchemy import func

# Posts: para cada categoría, los 3 más recientes
ranked = (
    select(
        Post,
        func.row_number().over(
            partition_by=Post.category_id,
            order_by=Post.published_at.desc(),
        ).label("rn"),
    )
    .where(Post.published == True)
    .subquery()
)

stmt = select(ranked).where(ranked.c.rn <= 3)

Búsqueda case-insensitive con index

# Asume idx_users_email_lower (módulo 3)
stmt = select(User).where(func.lower(User.email) == "maria@blog.local".lower())

Importante: la expresión debe coincidir con el index. func.lower(User.email) == "maria@blog.local".lower() matchea idx_users_email_lower.

Agregar tags como string (STRING_AGG)

from sqlalchemy import func

stmt = (
    select(
        Post.title,
        func.string_agg(Tag.name, ', ').label("tags"),
    )
    .join(Post.tags)
    .group_by(Post.id, Post.title)
)

Posts más comentados

stmt = (
    select(
        Post.title,
        func.count(Comment.id).label("num_comments"),
    )
    .outerjoin(Post.comments)
    .where(Post.published == True)
    .group_by(Post.id, Post.title)
    .order_by(func.count(Comment.id).desc())
    .limit(10)
)

Todas las queries del módulo 2 traducidas

Recordamos las 10 queries del producto del módulo 2 cápsula 08. Aquí algunas en SQLAlchemy:

Query 1: Feed principal con autor + category + tags

from sqlalchemy import func, select
from sqlalchemy.orm import selectinload  # cápsula 07

stmt = (
    select(Post)
    .options(
        selectinload(Post.author),
        selectinload(Post.category),
        selectinload(Post.tags),
    )
    .where(Post.published == True)
    .order_by(Post.published_at.desc())
    .limit(10)
)

posts = session.execute(stmt).scalars().all()

for p in posts:
    print(f"{p.title} — {p.author.username} — {p.category.name if p.category else 'Sin cat'}")
    print(f"  Tags: {', '.join(t.name for t in p.tags)}")

Query 3: Top 5 autores por posts publicados

stmt = (
    select(
        User.username,
        User.full_name,
        func.count(Post.id).label("posts_publicados"),
        func.max(Post.published_at).label("ultimo_post"),
    )
    .join(User.posts)
    .where(Post.published == True)
    .group_by(User.id, User.username, User.full_name)
    .order_by(func.count(Post.id).desc())
    .limit(5)
)

for username, full_name, count, last in session.execute(stmt):
    print(f"{username} ({full_name}): {count} posts, último: {last}")

Query 4: Top 10 tags

stmt = (
    select(
        Tag.name,
        func.count(post_tags_table.c.post_id).label("uso"),
    )
    .join(post_tags_table, Tag.id == post_tags_table.c.tag_id)
    .group_by(Tag.id, Tag.name)
    .order_by(func.count().desc())
    .limit(10)
)

Query 8: Posts relacionados (mismo tag)

post_x = session.execute(select(Post).where(Post.slug == "postgresql-produccion")).scalar_one()
post_x_tag_ids = select(post_tags_table.c.tag_id).where(post_tags_table.c.post_id == post_x.id)

stmt = (
    select(Post, func.count(post_tags_table.c.tag_id).label("tags_compartidos"))
    .join(post_tags_table, Post.id == post_tags_table.c.post_id)
    .where(
        post_tags_table.c.tag_id.in_(post_x_tag_ids),
        Post.id != post_x.id,
        Post.published == True,
    )
    .group_by(Post.id)
    .order_by(func.count().desc())
    .limit(5)
)

El blog completo en SQLAlchemy lo organizamos en la cápsula 08 con Repository pattern.

Ejecutar SQL crudo cuando sea necesario

A veces es más fácil escribir SQL directo. SQLAlchemy lo permite:

from sqlalchemy import text

stmt = text("""
    SELECT username, count(*) AS n
    FROM users u
    JOIN posts p ON p.author_id = u.id
    WHERE p.published = true
    GROUP BY username
    ORDER BY n DESC
    LIMIT :limit
""").bindparams(limit=5)

result = session.execute(stmt)
for row in result:
    print(row.username, row.n)

No abuses de text() — pierdes el tipado y el IDE no te ayuda. Úsalo solo cuando el ORM realmente no alcance.

Compose queries reutilizables

Las queries de SQLAlchemy son objetos. Puedes pasarlas como argumentos, modificarlas, reutilizarlas:

def published_posts_query():
    """Query base de posts publicados."""
    return select(Post).where(Post.published == True)

def by_author(stmt, username):
    """Filtra una query por autor."""
    return stmt.join(Post.author).where(User.username == username)

def order_recent(stmt):
    """Ordena por fecha descendente."""
    return stmt.order_by(Post.published_at.desc())

# Uso composable
stmt = order_recent(by_author(published_posts_query(), "maria")).limit(10)
posts = session.execute(stmt).scalars().all()

Esto NO es posible con text() — es una de las ventajas del ORM.

Errores comunes

MultipleResultsFound

session.execute(select(User)).scalar_one()
# ERROR si hay >1 user

Solución: usa scalars().first() o scalars().all() si esperas múltiples.

NoResultFound

session.execute(select(User).where(User.username == "noexiste")).scalar_one()
# ERROR si no hay match

Solución: scalar_one_or_none().

Olvidar .scalars()

result = session.execute(select(User)).all()
# Retorna [Row(User=<User>), ...] — tuplas con un elemento

result = session.execute(select(User)).scalars().all()
# Retorna [<User>, <User>, ...] — lista plana

== con None

# ❌
.where(User.deleted_at == None)
# Funciona pero es anti-patrón Python (linters quejan)

# ✅
.where(User.deleted_at.is_(None))

Ejercicios

Ejercicio 1. Implementa la query "feed principal" de SQL puro como select() con SQLAlchemy. Verifica que retorna los mismos resultados.

from sqlalchemy import select
from sqlalchemy.orm import selectinload
from app.database import db_session
from app.models import Post

with db_session() as session:
    posts = session.execute(
        select(Post)
        .options(selectinload(Post.author), selectinload(Post.category))
        .where(Post.published == True)
        .order_by(Post.published_at.desc())
        .limit(10)
    ).scalars().all()
    
    for p in posts:
        print(f"{p.title} — {p.author.username}")

Ejercicio 2. Cuenta cuántos posts y comments hay agrupados por autor.

from sqlalchemy import func, select
from app.database import db_session
from app.models import User, Post, Comment

with db_session() as session:
    stmt = (
        select(
            User.username,
            func.count(Post.id.distinct()).label("posts"),
            func.count(Comment.id.distinct()).label("comments"),
        )
        .outerjoin(Post, Post.author_id == User.id)
        .outerjoin(Comment, Comment.author_id == User.id)
        .group_by(User.id, User.username)
    )
    
    for u, p, c in session.execute(stmt):
        print(f"{u}: {p} posts, {c} comments")

Ejercicio 3. Encuentra usuarios que han comentado pero nunca publicado, usando EXISTS y NOT EXISTS.

from sqlalchemy import select, exists
from app.database import db_session
from app.models import User, Post, Comment

with db_session() as session:
    stmt = (
        select(User.username)
        .where(
            exists().where(Comment.author_id == User.id),
            ~exists().where(Post.author_id == User.id, Post.published == True),
        )
    )
    
    for (username,) in session.execute(stmt):
        print(username)

Ejercicio 4. Implementa "tags relacionados con postgresql" (co-ocurrencia, query 9 del módulo 2).

from sqlalchemy import func, select
from sqlalchemy.orm import aliased
from app.database import db_session
from app.models import Tag, post_tags_table

with db_session() as session:
    # Tag base
    tag_pg = session.execute(select(Tag).where(Tag.slug == "postgresql")).scalar_one()
    
    # Aliases para joinear post_tags consigo misma
    pt1 = aliased(post_tags_table)
    pt2 = aliased(post_tags_table)
    
    stmt = (
        select(
            Tag.name.label("relacionado"),
            func.count().label("co_ocurrencia"),
        )
        .select_from(pt1)
        .join(pt2, pt1.c.post_id == pt2.c.post_id)  # mismo post
        .join(Tag, Tag.id == pt2.c.tag_id)
        .where(pt1.c.tag_id == tag_pg.id, pt2.c.tag_id != tag_pg.id)
        .group_by(Tag.id, Tag.name)
        .order_by(func.count().desc())
    )
    
    for name, n in session.execute(stmt):
        print(f"{name}: {n}")

Ejercicio 5. Construye una función search_posts(session, query: str, tag: str | None = None) que retorna posts cuyo título contiene query (case-insensitive) y opcionalmente filtrados por tag.

from sqlalchemy import func, select
from sqlalchemy.orm import Session, selectinload
from app.models import Post, Tag

def search_posts(session: Session, query: str, tag: str | None = None) -> list[Post]:
    stmt = (
        select(Post)
        .options(selectinload(Post.author), selectinload(Post.tags))
        .where(
            Post.published == True,
            Post.title.ilike(f"%{query}%"),
        )
    )
    
    if tag:
        stmt = stmt.join(Post.tags).where(Tag.slug == tag)
    
    return list(session.execute(stmt).scalars().unique().all())

# Uso
from app.database import db_session
with db_session() as session:
    results = search_posts(session, "postgres", tag="tutorial")
    for p in results:
        print(p.title)

Resumen

• select(Model) o select(Model.col1, Model.col2) para construir queries
• .where() con ==, !=, <, .in_(), .like(), .is_(None), etc.
• .join(Model) o .join(Model.relationship) (preferido si hay relación)
• .outerjoin() para LEFT JOIN
• .order_by(), .limit(), .offset() para ordenamiento y paginación
• func.count(), func.sum(), func.avg(), func.X() para cualquier función SQL
• .group_by() + .having() para agrupaciones
• func.X(...).filter(condition) para FILTER WHERE
• .cte() para Common Table Expressions
• Resultados: .scalars().all() para listas de modelos, .execute().all() para tuplas
• Subqueries: .scalar_subquery() para escalares, select(...) directo en .in_()
• Composable: las queries son objetos Python, encadenables y reutilizables

En la siguiente cápsula, el tema más importante de performance: eager vs lazy loading y el N+1 problem.

Recursos Adicionales

1. SELECT Statement — SQLAlchemy 2.0
2. ORM Query Guide
3. func — SQL Function Construction
4. Window Functions

Siguiente: Cápsula 07 — Eager vs Lazy Loading.