Tutorial Flask: Consultas con Joins y filtros

Consultas con JOIN

Las consultas con JOIN en SQLAlchemy permiten combinar datos de múltiples tablas relacionadas en una sola consulta, optimizando el rendimiento y reduciendo el número de operaciones contra la base de datos. Esta técnica es fundamental cuando trabajamos con modelos que tienen relaciones definidas.

SQLAlchemy ofrece varios métodos para realizar joins explícitos e implícitos, cada uno con sus propias ventajas según el contexto de la consulta. Los joins implícitos aprovechan las relaciones ya definidas en los modelos, mientras que los explícitos nos dan mayor control sobre cómo se combinan las tablas.

Joins implícitos con relaciones

Cuando tenemos relaciones definidas entre modelos, SQLAlchemy puede realizar joins automáticamente al acceder a atributos relacionados. Consideremos estos modelos de ejemplo:

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import relationship

db = SQLAlchemy()

class Usuario(db.Model):
    __tablename__ = 'usuarios'
    
    id = db.Column(db.Integer, primary_key=True)
    nombre = db.Column(db.String(100), nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    
    # Relación con pedidos
    pedidos = relationship('Pedido', back_populates='usuario')

class Pedido(db.Model):
    __tablename__ = 'pedidos'
    
    id = db.Column(db.Integer, primary_key=True)
    total = db.Column(db.Decimal(10, 2), nullable=False)
    fecha = db.Column(db.DateTime, nullable=False)
    usuario_id = db.Column(db.Integer, db.ForeignKey('usuarios.id'))
    
    # Relación con usuario
    usuario = relationship('Usuario', back_populates='pedidos')

Para consultar usuarios junto con sus pedidos usando joins implícitos:

# Join implícito - SQLAlchemy detecta la relación automáticamente
usuarios_con_pedidos = db.session.query(Usuario).join(Usuario.pedidos).all()

# Filtrar usuarios que tienen pedidos con total mayor a 100
usuarios_grandes_compras = (
    db.session.query(Usuario)
    .join(Usuario.pedidos)
    .filter(Pedido.total > 100)
    .all()
)

Joins explícitos con control manual

Los joins explícitos nos permiten especificar exactamente cómo queremos combinar las tablas, ofreciendo mayor flexibilidad:

# Join explícito especificando las tablas y condición
resultado = (
    db.session.query(Usuario, Pedido)
    .join(Pedido, Usuario.id == Pedido.usuario_id)
    .filter(Pedido.total > 50)
    .all()
)

# Join explícito con alias para consultas más complejas
from sqlalchemy.orm import aliased

PedidoAlias = aliased(Pedido)
usuarios_multiples_pedidos = (
    db.session.query(Usuario)
    .join(PedidoAlias, Usuario.id == PedidoAlias.usuario_id)
    .group_by(Usuario.id)
    .having(db.func.count(PedidoAlias.id) > 1)
    .all()
)

Tipos de JOIN disponibles

SQLAlchemy soporta diferentes tipos de join según las necesidades de la consulta:

Inner Join (por defecto):

# Solo usuarios que tienen al menos un pedido
usuarios_con_pedidos = db.session.query(Usuario).join(Pedido).all()

Left Join (outer join):

# Todos los usuarios, incluso los que no tienen pedidos
todos_usuarios = db.session.query(Usuario).outerjoin(Pedido).all()

# Usuarios sin pedidos
usuarios_sin_pedidos = (
    db.session.query(Usuario)
    .outerjoin(Pedido)
    .filter(Pedido.id.is_(None))
    .all()
)

Consultas con múltiples joins

Para consultas complejas que involucran múltiples tablas, podemos encadenar varios joins:

class Categoria(db.Model):
    __tablename__ = 'categorias'
    
    id = db.Column(db.Integer, primary_key=True)
    nombre = db.Column(db.String(50), nullable=False)

class Producto(db.Model):
    __tablename__ = 'productos'
    
    id = db.Column(db.Integer, primary_key=True)
    nombre = db.Column(db.String(100), nullable=False)
    precio = db.Column(db.Decimal(10, 2), nullable=False)
    categoria_id = db.Column(db.Integer, db.ForeignKey('categorias.id'))
    
    categoria = relationship('Categoria')

class DetallePedido(db.Model):
    __tablename__ = 'detalles_pedido'
    
    id = db.Column(db.Integer, primary_key=True)
    pedido_id = db.Column(db.Integer, db.ForeignKey('pedidos.id'))
    producto_id = db.Column(db.Integer, db.ForeignKey('productos.id'))
    cantidad = db.Column(db.Integer, nullable=False)
    
    pedido = relationship('Pedido')
    producto = relationship('Producto')

Consulta que combina múltiples tablas para obtener información detallada:

# Usuarios con sus pedidos, productos y categorías
consulta_completa = (
    db.session.query(
        Usuario.nombre.label('usuario'),
        Pedido.fecha.label('fecha_pedido'),
        Producto.nombre.label('producto'),
        Categoria.nombre.label('categoria'),
        DetallePedido.cantidad
    )
    .join(Pedido, Usuario.id == Pedido.usuario_id)
    .join(DetallePedido, Pedido.id == DetallePedido.pedido_id)
    .join(Producto, DetallePedido.producto_id == Producto.id)
    .join(Categoria, Producto.categoria_id == Categoria.id)
    .filter(Categoria.nombre == 'Electrónicos')
    .all()
)

Optimización con eager loading

Para evitar el problema N+1 al acceder a relaciones, podemos usar joinedload para cargar datos relacionados en una sola consulta:

from sqlalchemy.orm import joinedload

# Carga usuarios con sus pedidos en una sola consulta
usuarios_optimizado = (
    db.session.query(Usuario)
    .options(joinedload(Usuario.pedidos))
    .all()
)

# Carga múltiples niveles de relaciones
usuarios_completo = (
    db.session.query(Usuario)
    .options(
        joinedload(Usuario.pedidos)
        .joinedload(Pedido.detalles)
        .joinedload(DetallePedido.producto)
    )
    .all()
)

Agregaciones con joins

Los joins combinados con funciones de agregación permiten obtener estadísticas útiles:

# Total gastado por cada usuario
gastos_por_usuario = (
    db.session.query(
        Usuario.nombre,
        db.func.sum(Pedido.total).label('total_gastado'),
        db.func.count(Pedido.id).label('num_pedidos')
    )
    .join(Pedido)
    .group_by(Usuario.id, Usuario.nombre)
    .order_by(db.func.sum(Pedido.total).desc())
    .all()
)

# Productos más vendidos por categoría
productos_populares = (
    db.session.query(
        Categoria.nombre.label('categoria'),
        Producto.nombre.label('producto'),
        db.func.sum(DetallePedido.cantidad).label('total_vendido')
    )
    .join(Producto, Categoria.id == Producto.categoria_id)
    .join(DetallePedido, Producto.id == DetallePedido.producto_id)
    .group_by(Categoria.id, Producto.id)
    .order_by(db.func.sum(DetallePedido.cantidad).desc())
    .limit(10)
    .all()
)

Filtros con WHERE

Los filtros con WHERE en SQLAlchemy nos permiten especificar condiciones precisas para seleccionar únicamente los registros que cumplan ciertos criterios. Esta funcionalidad es esencial para construir consultas eficientes que devuelvan exactamente los datos que necesitamos, especialmente cuando trabajamos con grandes volúmenes de información.

SQLAlchemy proporciona una sintaxis intuitiva para construir condiciones complejas usando el método filter() y filter_by(), cada uno con sus propias ventajas según el tipo de filtrado que necesitemos realizar.

Filtros básicos con filter_by()

El método filter_by() es la forma más directa de filtrar por valores exactos en columnas específicas:

# Buscar usuarios por email específico
usuario = db.session.query(Usuario).filter_by(email='juan@ejemplo.com').first()

# Filtrar pedidos por usuario específico
pedidos_usuario = db.session.query(Pedido).filter_by(usuario_id=1).all()

# Combinar múltiples condiciones con filter_by
pedidos_recientes = (
    db.session.query(Pedido)
    .filter_by(usuario_id=1)
    .filter_by(estado='completado')
    .all()
)

Filtros avanzados con filter()

El método filter() ofrece mayor flexibilidad para construir condiciones complejas usando operadores de comparación:

# Pedidos con total mayor a 100
pedidos_grandes = db.session.query(Pedido).filter(Pedido.total > 100).all()

# Usuarios cuyo nombre contiene una cadena específica
usuarios_juan = db.session.query(Usuario).filter(Usuario.nombre.like('%Juan%')).all()

# Pedidos dentro de un rango de fechas
from datetime import datetime, timedelta

fecha_inicio = datetime.now() - timedelta(days=30)
fecha_fin = datetime.now()

pedidos_mes = (
    db.session.query(Pedido)
    .filter(Pedido.fecha >= fecha_inicio)
    .filter(Pedido.fecha <= fecha_fin)
    .all()
)

Operadores de comparación

SQLAlchemy soporta una amplia gama de operadores de comparación para diferentes tipos de filtrado:

Operadores numéricos:

# Mayor que, menor que, igual
productos_caros = db.session.query(Producto).filter(Producto.precio > 50).all()
productos_baratos = db.session.query(Producto).filter(Producto.precio <= 20).all()
producto_exacto = db.session.query(Producto).filter(Producto.precio == 15.99).first()

# Diferente de
productos_no_gratis = db.session.query(Producto).filter(Producto.precio != 0).all()

Operadores de texto:

# Búsqueda con patrones usando LIKE
usuarios_gmail = db.session.query(Usuario).filter(Usuario.email.like('%@gmail.com')).all()

# Búsqueda insensible a mayúsculas con ILIKE
usuarios_juan = db.session.query(Usuario).filter(Usuario.nombre.ilike('%juan%')).all()

# Coincidencia exacta al inicio o final
productos_tech = db.session.query(Producto).filter(Producto.nombre.like('Tech%')).all()

Filtros con listas y rangos

Para filtrar usando múltiples valores o rangos, SQLAlchemy proporciona métodos específicos:

# Filtrar por lista de valores con in_()
ids_usuarios = [1, 3, 5, 7]
usuarios_seleccionados = db.session.query(Usuario).filter(Usuario.id.in_(ids_usuarios)).all()

# Filtrar por rango con between()
productos_rango = (
    db.session.query(Producto)
    .filter(Producto.precio.between(10, 50))
    .all()
)

# Excluir valores específicos con notin_()
categorias_excluidas = ['Descontinuado', 'Agotado']
productos_disponibles = (
    db.session.query(Producto)
    .join(Categoria)
    .filter(Categoria.nombre.notin_(categorias_excluidas))
    .all()
)

Filtros con valores nulos

El manejo de valores nulos requiere métodos específicos para garantizar consultas precisas:

# Registros con valores nulos
usuarios_sin_telefono = db.session.query(Usuario).filter(Usuario.telefono.is_(None)).all()

# Registros con valores no nulos
usuarios_con_telefono = db.session.query(Usuario).filter(Usuario.telefono.isnot(None)).all()

# Combinar con otros filtros
usuarios_activos_completos = (
    db.session.query(Usuario)
    .filter(Usuario.activo == True)
    .filter(Usuario.email.isnot(None))
    .filter(Usuario.telefono.isnot(None))
    .all()
)

Operadores lógicos complejos

Para construir condiciones complejas, podemos combinar múltiples filtros usando operadores lógicos:

from sqlalchemy import and_, or_, not_

# Operador AND explícito
pedidos_grandes_recientes = (
    db.session.query(Pedido)
    .filter(and_(
        Pedido.total > 100,
        Pedido.fecha >= datetime.now() - timedelta(days=7)
    ))
    .all()
)

# Operador OR para múltiples condiciones
usuarios_especiales = (
    db.session.query(Usuario)
    .filter(or_(
        Usuario.email.like('%@empresa.com'),
        Usuario.tipo == 'premium',
        Usuario.fecha_registro < datetime(2020, 1, 1)
    ))
    .all()
)

# Operador NOT para exclusiones
productos_normales = (
    db.session.query(Producto)
    .filter(not_(or_(
        Producto.precio > 1000,
        Producto.categoria_id.in_([1, 2])  # Categorías premium
    )))
    .all()
)

Filtros en consultas con JOIN

Los filtros se pueden aplicar tanto a la tabla principal como a las tablas relacionadas en consultas con JOIN:

# Filtrar por condiciones en tabla relacionada
usuarios_compras_grandes = (
    db.session.query(Usuario)
    .join(Pedido)
    .filter(Pedido.total > 200)
    .distinct()
    .all()
)

# Múltiples filtros en diferentes tablas
consulta_compleja = (
    db.session.query(Usuario, Pedido, Producto)
    .join(Pedido, Usuario.id == Pedido.usuario_id)
    .join(DetallePedido, Pedido.id == DetallePedido.pedido_id)
    .join(Producto, DetallePedido.producto_id == Producto.id)
    .filter(Usuario.activo == True)  # Filtro en tabla usuarios
    .filter(Pedido.fecha >= datetime(2024, 1, 1))  # Filtro en tabla pedidos
    .filter(Producto.precio > 25)  # Filtro en tabla productos
    .all()
)

Filtros con subconsultas

Las subconsultas permiten crear filtros basados en resultados de otras consultas:

# Usuarios que han realizado más de 3 pedidos
subquery = (
    db.session.query(Pedido.usuario_id)
    .group_by(Pedido.usuario_id)
    .having(db.func.count(Pedido.id) > 3)
    .subquery()
)

usuarios_frecuentes = (
    db.session.query(Usuario)
    .filter(Usuario.id.in_(subquery))
    .all()
)

# Productos nunca vendidos
productos_sin_ventas = (
    db.session.query(Producto)
    .filter(~Producto.id.in_(
        db.session.query(DetallePedido.producto_id).distinct()
    ))
    .all()
)

Filtros con funciones de agregación

Cuando usamos funciones de agregación, los filtros se aplican con having() en lugar de filter():

# Categorías con más de 5 productos
categorias_populares = (
    db.session.query(
        Categoria.nombre,
        db.func.count(Producto.id).label('total_productos')
    )
    .join(Producto)
    .group_by(Categoria.id, Categoria.nombre)
    .having(db.func.count(Producto.id) > 5)
    .all()
)

# Usuarios con gasto total superior a 500
grandes_clientes = (
    db.session.query(
        Usuario.nombre,
        db.func.sum(Pedido.total).label('gasto_total')
    )
    .join(Pedido)
    .group_by(Usuario.id, Usuario.nombre)
    .having(db.func.sum(Pedido.total) > 500)
    .order_by(db.func.sum(Pedido.total).desc())
    .all()
)

Optimización de filtros

Para mejorar el rendimiento de las consultas con filtros, es importante considerar el orden y la selectividad:

# Aplicar filtros más selectivos primero
consulta_optimizada = (
    db.session.query(Pedido)
    .filter(Pedido.estado == 'completado')  # Filtro más selectivo primero
    .filter(Pedido.fecha >= fecha_inicio)   # Filtro menos selectivo después
    .filter(Pedido.total > 50)
    .all()
)

# Usar índices de base de datos para filtros frecuentes
# En el modelo, agregar índices para columnas filtradas frecuentemente
class Pedido(db.Model):
    __tablename__ = 'pedidos'
    
    id = db.Column(db.Integer, primary_key=True)
    estado = db.Column(db.String(20), index=True)  # Índice para filtros frecuentes
    fecha = db.Column(db.DateTime, index=True)     # Índice para rangos de fechas
    total = db.Column(db.Decimal(10, 2))