Consultas con Joins y filtros

Filtros avanzados con where(), like(), in_()

Los filtros básicos que hemos visto hasta ahora nos permiten buscar registros por valores exactos, pero en aplicaciones reales necesitamos criterios de búsqueda más flexibles. SQLAlchemy proporciona métodos específicos para crear consultas sofisticadas que se adapten a diferentes necesidades de filtrado.

Filtrado con condiciones múltiples usando where()

El método where() nos permite combinar múltiples condiciones de filtrado de forma clara y legible. A diferencia del método filter(), where() ofrece una sintaxis más moderna y expresiva.

from sqlalchemy.orm import Session
from sqlalchemy import and_, or_

def obtener_usuarios_activos_por_edad(db: Session, edad_minima: int, edad_maxima: int):
    return db.query(Usuario).where(
        and_(
            Usuario.activo == True,
            Usuario.edad >= edad_minima,
            Usuario.edad <= edad_maxima
        )
    ).all()

Para condiciones alternativas, utilizamos el operador or_():

def buscar_usuarios_por_criterios(db: Session, nombre: str = None, email: str = None):
    condiciones = []
    
    if nombre:
        condiciones.append(Usuario.nombre.ilike(f"%{nombre}%"))
    if email:
        condiciones.append(Usuario.email.ilike(f"%{email}%"))
    
    if condiciones:
        return db.query(Usuario).where(or_(*condiciones)).all()
    
    return db.query(Usuario).all()

Búsquedas de texto con like() e ilike()

Las búsquedas de texto parcial son fundamentales para implementar funcionalidades de búsqueda en aplicaciones web. SQLAlchemy ofrece varios métodos para realizar estas consultas:

Búsqueda sensible a mayúsculas con like():

def buscar_productos_por_nombre(db: Session, termino: str):
    # Busca productos que contengan el término exacto
    return db.query(Producto).where(
        Producto.nombre.like(f"%{termino}%")
    ).all()

def buscar_productos_que_empiecen_con(db: Session, prefijo: str):
    # Busca productos cuyo nombre empiece con el prefijo
    return db.query(Producto).where(
        Producto.nombre.like(f"{prefijo}%")
    ).all()

Búsqueda insensible a mayúsculas con ilike():

def buscar_usuarios_flexible(db: Session, termino_busqueda: str):
    # Busca en nombre y email sin importar mayúsculas/minúsculas
    return db.query(Usuario).where(
        or_(
            Usuario.nombre.ilike(f"%{termino_busqueda}%"),
            Usuario.email.ilike(f"%{termino_busqueda}%")
        )
    ).all()

Filtrado por listas de valores con in_()

El método in_() resulta especialmente útil cuando necesitamos filtrar registros que coincidan con cualquier valor de una lista predefinida. Esto es común en filtros de categorías, estados o identificadores múltiples.

def obtener_productos_por_categorias(db: Session, categorias: list[str]):
    return db.query(Producto).where(
        Producto.categoria.in_(categorias)
    ).all()

def obtener_usuarios_por_ids(db: Session, ids_usuarios: list[int]):
    return db.query(Usuario).where(
        Usuario.id.in_(ids_usuarios)
    ).all()

Para excluir valores de una lista, combinamos in_() con el operador de negación:

def obtener_productos_excepto_categorias(db: Session, categorias_excluidas: list[str]):
    return db.query(Producto).where(
        ~Producto.categoria.in_(categorias_excluidas)
    ).all()

Implementación en endpoints de FastAPI

Estos filtros se integran naturalmente en nuestros endpoints de FastAPI para crear APIs de búsqueda flexibles:

from fastapi import FastAPI, Depends, Query
from typing import Optional

@app.get("/usuarios/buscar")
def buscar_usuarios(
    nombre: Optional[str] = Query(None, description="Buscar por nombre"),
    edad_min: Optional[int] = Query(None, description="Edad mínima"),
    edad_max: Optional[int] = Query(None, description="Edad máxima"),
    activo: Optional[bool] = Query(None, description="Estado activo"),
    db: Session = Depends(get_db)
):
    query = db.query(Usuario)
    
    if nombre:
        query = query.where(Usuario.nombre.ilike(f"%{nombre}%"))
    
    if edad_min is not None:
        query = query.where(Usuario.edad >= edad_min)
    
    if edad_max is not None:
        query = query.where(Usuario.edad <= edad_max)
    
    if activo is not None:
        query = query.where(Usuario.activo == activo)
    
    return query.all()

Un endpoint más avanzado que combina múltiples tipos de filtros:

@app.get("/productos/filtrar")
def filtrar_productos(
    busqueda: Optional[str] = Query(None, description="Término de búsqueda"),
    categorias: Optional[str] = Query(None, description="Categorías separadas por comas"),
    precio_min: Optional[float] = Query(None, description="Precio mínimo"),
    precio_max: Optional[float] = Query(None, description="Precio máximo"),
    db: Session = Depends(get_db)
):
    query = db.query(Producto)
    
    # Búsqueda de texto en nombre y descripción
    if busqueda:
        query = query.where(
            or_(
                Producto.nombre.ilike(f"%{busqueda}%"),
                Producto.descripcion.ilike(f"%{busqueda}%")
            )
        )
    
    # Filtro por categorías múltiples
    if categorias:
        lista_categorias = [cat.strip() for cat in categorias.split(",")]
        query = query.where(Producto.categoria.in_(lista_categorias))
    
    # Rango de precios
    if precio_min is not None:
        query = query.where(Producto.precio >= precio_min)
    
    if precio_max is not None:
        query = query.where(Producto.precio <= precio_max)
    
    return query.all()

Estos filtros avanzados nos permiten crear consultas precisas y eficientes que se adaptan a las necesidades específicas de búsqueda de nuestros usuarios, proporcionando una base sólida para implementar funcionalidades de filtrado en aplicaciones web profesionales.

Joins básicos entre tablas

Cuando trabajamos con bases de datos relacionales, es común que la información esté distribuida en múltiples tablas conectadas entre sí. Los joins nos permiten combinar datos de diferentes tablas en una sola consulta, evitando la necesidad de realizar múltiples consultas separadas.

Definiendo relaciones entre modelos

Antes de realizar joins, necesitamos establecer las relaciones entre nuestros modelos SQLAlchemy. Esto se hace mediante claves foráneas y el uso de relationship():

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from database import Base

class Usuario(Base):
    __tablename__ = "usuarios"
    
    id = Column(Integer, primary_key=True, index=True)
    nombre = Column(String, index=True)
    email = Column(String, unique=True, index=True)
    
    # Relación uno a muchos con pedidos
    pedidos = relationship("Pedido", back_populates="usuario")

class Pedido(Base):
    __tablename__ = "pedidos"
    
    id = Column(Integer, primary_key=True, index=True)
    producto = Column(String)
    cantidad = Column(Integer)
    usuario_id = Column(Integer, ForeignKey("usuarios.id"))
    
    # Relación muchos a uno con usuario
    usuario = relationship("Usuario", back_populates="pedidos")

Inner Join - Combinando datos relacionados

El inner join es el tipo de join más común y devuelve únicamente los registros que tienen coincidencias en ambas tablas. Es ideal cuando necesitamos datos que existen en ambas tablas relacionadas:

from sqlalchemy.orm import Session

def obtener_pedidos_con_usuarios(db: Session):
    return db.query(Pedido, Usuario).join(Usuario).all()

def obtener_usuarios_con_pedidos(db: Session):
    # Solo usuarios que tienen al menos un pedido
    return db.query(Usuario).join(Pedido).all()

Para acceder a campos específicos de ambas tablas:

def obtener_resumen_pedidos(db: Session):
    return db.query(
        Usuario.nombre,
        Usuario.email,
        Pedido.producto,
        Pedido.cantidad
    ).join(Usuario).all()

Left Join - Incluyendo registros sin coincidencias

El left join incluye todos los registros de la tabla principal, incluso si no tienen coincidencias en la tabla relacionada. Esto es útil cuando queremos obtener todos los registros de una tabla, independientemente de si tienen datos relacionados:

def obtener_todos_usuarios_con_pedidos(db: Session):
    # Incluye usuarios sin pedidos (pedidos aparecerán como None)
    return db.query(Usuario, Pedido).outerjoin(Pedido).all()

def contar_pedidos_por_usuario(db: Session):
    from sqlalchemy import func
    
    return db.query(
        Usuario.nombre,
        func.count(Pedido.id).label("total_pedidos")
    ).outerjoin(Pedido).group_by(Usuario.id, Usuario.nombre).all()

Joins con filtros combinados

Los joins se pueden combinar con los filtros que hemos aprendido anteriormente para crear consultas más específicas:

def buscar_pedidos_por_usuario_y_producto(db: Session, nombre_usuario: str, producto: str):
    return db.query(Pedido, Usuario).join(Usuario).where(
        and_(
            Usuario.nombre.ilike(f"%{nombre_usuario}%"),
            Pedido.producto.ilike(f"%{producto}%")
        )
    ).all()

def obtener_usuarios_con_pedidos_multiples(db: Session, productos: list[str]):
    return db.query(Usuario).join(Pedido).where(
        Pedido.producto.in_(productos)
    ).distinct().all()

Implementación en endpoints de FastAPI

Los joins se integran perfectamente en nuestros endpoints de FastAPI para proporcionar datos enriquecidos a nuestros clientes:

from fastapi import FastAPI, Depends
from typing import List

@app.get("/usuarios/{usuario_id}/pedidos")
def obtener_pedidos_usuario(usuario_id: int, db: Session = Depends(get_db)):
    return db.query(Pedido).join(Usuario).where(Usuario.id == usuario_id).all()

@app.get("/pedidos/detallados")
def obtener_pedidos_detallados(db: Session = Depends(get_db)):
    resultados = db.query(
        Pedido.id,
        Pedido.producto,
        Pedido.cantidad,
        Usuario.nombre.label("nombre_usuario"),
        Usuario.email.label("email_usuario")
    ).join(Usuario).all()
    
    return [
        {
            "pedido_id": resultado.id,
            "producto": resultado.producto,
            "cantidad": resultado.cantidad,
            "usuario": {
                "nombre": resultado.nombre_usuario,
                "email": resultado.email_usuario
            }
        }
        for resultado in resultados
    ]

Un endpoint más complejo que combina joins con filtros avanzados:

@app.get("/reportes/usuarios-activos")
def obtener_reporte_usuarios_activos(
    producto_filtro: Optional[str] = Query(None),
    cantidad_minima: Optional[int] = Query(None),
    db: Session = Depends(get_db)
):
    query = db.query(
        Usuario.nombre,
        Usuario.email,
        func.count(Pedido.id).label("total_pedidos"),
        func.sum(Pedido.cantidad).label("cantidad_total")
    ).outerjoin(Pedido)
    
    # Aplicar filtros si se proporcionan
    if producto_filtro:
        query = query.where(Pedido.producto.ilike(f"%{producto_filtro}%"))
    
    if cantidad_minima:
        query = query.where(Pedido.cantidad >= cantidad_minima)
    
    return query.group_by(Usuario.id, Usuario.nombre, Usuario.email).all()

Trabajando con múltiples tablas

Cuando necesitamos datos de tres o más tablas, podemos encadenar múltiples joins:

# Modelo adicional para el ejemplo
class Categoria(Base):
    __tablename__ = "categorias"
    
    id = Column(Integer, primary_key=True)
    nombre = Column(String)
    
    productos = relationship("ProductoCompleto", back_populates="categoria")

class ProductoCompleto(Base):
    __tablename__ = "productos_completos"
    
    id = Column(Integer, primary_key=True)
    nombre = Column(String)
    categoria_id = Column(Integer, ForeignKey("categorias.id"))
    
    categoria = relationship("Categoria", back_populates="productos")

# Consulta con múltiples joins
def obtener_pedidos_completos(db: Session):
    return db.query(
        Usuario.nombre.label("usuario"),
        ProductoCompleto.nombre.label("producto"),
        Categoria.nombre.label("categoria"),
        Pedido.cantidad
    ).join(Usuario).join(ProductoCompleto).join(Categoria).all()

Los joins básicos nos proporcionan la capacidad de crear consultas que aprovechan las relaciones entre tablas, permitiendo obtener información completa y contextualizada en una sola operación de base de datos, lo que resulta más eficiente que realizar múltiples consultas separadas.