Operaciones CRUD y consultas

Operaciones CRUD básicas

Las operaciones CRUD (Create, Read, Update, Delete) constituyen el núcleo de cualquier aplicación que gestione datos. En Flask con SQLAlchemy, estas operaciones se realizan a través de métodos específicos que interactúan directamente con la base de datos, proporcionando una interfaz intuitiva para manipular los registros de nuestros modelos.

Creación de registros (Create)

La creación de nuevos registros en SQLAlchemy sigue un patrón consistente: instanciar el modelo, configurar sus atributos y persistir los cambios en la base de datos.

from app import db
from models import Usuario

# Crear una nueva instancia del modelo
nuevo_usuario = Usuario(
    nombre="Ana García",
    email="ana@ejemplo.com",
    edad=28
)

# Agregar al contexto de la sesión
db.session.add(nuevo_usuario)

# Confirmar los cambios en la base de datos
db.session.commit()

Para múltiples registros, SQLAlchemy ofrece el método add_all() que optimiza las operaciones de inserción:

usuarios = [
    Usuario(nombre="Carlos López", email="carlos@ejemplo.com", edad=32),
    Usuario(nombre="María Ruiz", email="maria@ejemplo.com", edad=25),
    Usuario(nombre="Pedro Sánchez", email="pedro@ejemplo.com", edad=29)
]

db.session.add_all(usuarios)
db.session.commit()

Lectura de registros (Read)

Las consultas de lectura en SQLAlchemy se construyen utilizando el atributo query del modelo, que proporciona una interfaz fluida para filtrar y recuperar datos.

Obtener todos los registros:

# Recuperar todos los usuarios
usuarios = Usuario.query.all()

# Iterar sobre los resultados
for usuario in usuarios:
    print(f"{usuario.nombre} - {usuario.email}")

Búsqueda por clave primaria:

# Buscar por ID (clave primaria)
usuario = Usuario.query.get(1)

if usuario:
    print(f"Usuario encontrado: {usuario.nombre}")
else:
    print("Usuario no encontrado")

Filtrado con condiciones:

# Filtrar por un campo específico
usuarios_jovenes = Usuario.query.filter(Usuario.edad < 30).all()

# Filtrar por múltiples condiciones
usuario_especifico = Usuario.query.filter(
    Usuario.nombre == "Ana García",
    Usuario.edad > 25
).first()

# Búsqueda por email (único)
usuario = Usuario.query.filter_by(email="ana@ejemplo.com").first()

Métodos de recuperación específicos:

# first() - Primer resultado o None
primer_usuario = Usuario.query.first()

# first_or_404() - Primer resultado o error 404
usuario = Usuario.query.filter_by(email="inexistente@ejemplo.com").first_or_404()

# count() - Contar registros
total_usuarios = Usuario.query.count()
usuarios_mayores = Usuario.query.filter(Usuario.edad >= 30).count()

Actualización de registros (Update)

La modificación de registros existentes requiere primero recuperar el objeto, modificar sus atributos y confirmar los cambios.

Actualización de un registro individual:

# Buscar el usuario a actualizar
usuario = Usuario.query.get(1)

if usuario:
    # Modificar los atributos
    usuario.nombre = "Ana García Martínez"
    usuario.edad = 29
    
    # Confirmar los cambios
    db.session.commit()
    print("Usuario actualizado correctamente")

Actualización masiva con filtros:

# Actualizar múltiples registros que cumplan una condición
usuarios_actualizados = Usuario.query.filter(Usuario.edad < 25).update({
    'activo': True
})

db.session.commit()
print(f"Se actualizaron {usuarios_actualizados} usuarios")

Actualización condicional:

usuario = Usuario.query.filter_by(email="ana@ejemplo.com").first()

if usuario:
    # Actualizar solo si se cumplen ciertas condiciones
    if usuario.edad < 30:
        usuario.categoria = "joven"
    else:
        usuario.categoria = "adulto"
    
    db.session.commit()

Eliminación de registros (Delete)

Las operaciones de eliminación en SQLAlchemy pueden realizarse sobre instancias específicas o mediante filtros que afecten múltiples registros.

Eliminación de un registro específico:

# Buscar y eliminar por ID
usuario = Usuario.query.get(1)

if usuario:
    db.session.delete(usuario)
    db.session.commit()
    print("Usuario eliminado correctamente")

Eliminación con filtros:

# Eliminar usuarios que cumplan una condición
usuarios_eliminados = Usuario.query.filter(Usuario.edad > 65).delete()

db.session.commit()
print(f"Se eliminaron {usuarios_eliminados} usuarios")

Eliminación por criterio específico:

# Eliminar usuario por email
usuario = Usuario.query.filter_by(email="usuario@eliminar.com").first()

if usuario:
    db.session.delete(usuario)
    db.session.commit()

Manejo de transacciones y errores

Las transacciones en SQLAlchemy garantizan la integridad de los datos mediante el uso de commit() y rollback():

try:
    # Operaciones múltiples en una transacción
    usuario1 = Usuario(nombre="Test 1", email="test1@ejemplo.com")
    usuario2 = Usuario(nombre="Test 2", email="test2@ejemplo.com")
    
    db.session.add(usuario1)
    db.session.add(usuario2)
    
    # Confirmar todas las operaciones
    db.session.commit()
    
except Exception as e:
    # Revertir cambios en caso de error
    db.session.rollback()
    print(f"Error en la transacción: {e}")

Verificación antes de operaciones:

# Verificar existencia antes de crear
email_existente = Usuario.query.filter_by(email="nuevo@ejemplo.com").first()

if not email_existente:
    nuevo_usuario = Usuario(
        nombre="Usuario Nuevo",
        email="nuevo@ejemplo.com",
        edad=30
    )
    db.session.add(nuevo_usuario)
    db.session.commit()
else:
    print("El email ya está registrado")

Estas operaciones CRUD forman la base de cualquier aplicación Flask que maneje datos persistentes. La combinación de métodos de consulta con las capacidades de transacción de SQLAlchemy proporciona un sistema robusto para gestionar la información de manera segura y eficiente.

Query y session

El objeto Query y la gestión de sesiones en SQLAlchemy representan los mecanismos fundamentales que controlan cómo interactuamos con la base de datos. Mientras que las operaciones CRUD básicas nos permiten manipular datos, el sistema de consultas y sesiones nos proporciona un control granular sobre el rendimiento, la consistencia y el comportamiento de nuestras aplicaciones.

El objeto Query y sus métodos avanzados

El objeto Query ofrece una interfaz rica para construir consultas complejas que van más allá de las operaciones básicas de filtrado. Estos métodos nos permiten optimizar el rendimiento y obtener exactamente los datos que necesitamos.

Ordenación y limitación de resultados:

# Ordenar por un campo específico
usuarios_ordenados = Usuario.query.order_by(Usuario.nombre).all()

# Ordenación descendente
usuarios_por_edad = Usuario.query.order_by(Usuario.edad.desc()).all()

# Limitar número de resultados
primeros_cinco = Usuario.query.limit(5).all()

# Paginación con offset
usuarios_pagina_2 = Usuario.query.offset(10).limit(5).all()

Métodos de agregación y estadísticas:

from sqlalchemy import func

# Contar registros con condiciones
usuarios_activos = Usuario.query.filter(Usuario.activo == True).count()

# Funciones de agregación
edad_promedio = db.session.query(func.avg(Usuario.edad)).scalar()
edad_maxima = db.session.query(func.max(Usuario.edad)).scalar()
edad_minima = db.session.query(func.min(Usuario.edad)).scalar()

# Agrupar resultados
usuarios_por_edad = db.session.query(
    Usuario.edad, 
    func.count(Usuario.id)
).group_by(Usuario.edad).all()

Consultas con múltiples condiciones:

# Operadores lógicos complejos
from sqlalchemy import and_, or_, not_

usuarios_filtrados = Usuario.query.filter(
    and_(
        Usuario.edad >= 18,
        or_(
            Usuario.ciudad == "Madrid",
            Usuario.ciudad == "Barcelona"
        )
    )
).all()

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

# Búsqueda en listas de valores
ciudades_objetivo = ["Madrid", "Barcelona", "Valencia"]
usuarios_ciudades = Usuario.query.filter(
    Usuario.ciudad.in_(ciudades_objetivo)
).all()

Gestión avanzada de sesiones

La sesión de SQLAlchemy actúa como una unidad de trabajo que mantiene el estado de los objetos y coordina las operaciones con la base de datos. Comprender su funcionamiento es crucial para desarrollar aplicaciones eficientes y confiables.

Estados de los objetos en la sesión:

# Crear un nuevo objeto (estado transient)
nuevo_usuario = Usuario(nombre="Test", email="test@ejemplo.com")
print(f"Estado inicial: {nuevo_usuario in db.session}")  # False

# Agregar a la sesión (estado pending)
db.session.add(nuevo_usuario)
print(f"En sesión: {nuevo_usuario in db.session}")  # True
print(f"ID antes de commit: {nuevo_usuario.id}")  # None

# Confirmar cambios (estado persistent)
db.session.commit()
print(f"ID después de commit: {nuevo_usuario.id}")  # Valor asignado

Manejo de la caché de sesión:

# La sesión mantiene una caché de objetos
usuario_1 = Usuario.query.get(1)
usuario_2 = Usuario.query.get(1)

# Ambas variables apuntan al mismo objeto en memoria
print(usuario_1 is usuario_2)  # True

# Refrescar objeto desde la base de datos
db.session.refresh(usuario_1)

# Expulsar objeto de la sesión
db.session.expunge(usuario_1)
print(f"Usuario en sesión: {usuario_1 in db.session}")  # False

Control de transacciones avanzado:

# Crear un punto de guardado (savepoint)
try:
    # Operación principal
    usuario = Usuario(nombre="Principal", email="principal@ejemplo.com")
    db.session.add(usuario)
    
    # Crear savepoint
    savepoint = db.session.begin_nested()
    
    try:
        # Operación secundaria que puede fallar
        usuario_duplicado = Usuario(nombre="Duplicado", email="principal@ejemplo.com")
        db.session.add(usuario_duplicado)
        db.session.commit()
        
    except Exception:
        # Revertir solo hasta el savepoint
        savepoint.rollback()
        print("Error en operación secundaria, continuando...")
    
    # La operación principal se mantiene
    db.session.commit()
    
except Exception as e:
    # Revertir toda la transacción
    db.session.rollback()
    print(f"Error general: {e}")

Optimización de consultas y rendimiento

El rendimiento de las consultas puede mejorarse significativamente mediante técnicas específicas que reducen el número de accesos a la base de datos y optimizan la transferencia de datos.

Carga eager vs lazy:

# Problema N+1: cada usuario genera una consulta adicional
usuarios = Usuario.query.all()
for usuario in usuarios:
    print(f"{usuario.nombre} tiene {len(usuario.pedidos)} pedidos")  # N consultas

# Solución con joinedload (eager loading)
from sqlalchemy.orm import joinedload

usuarios_con_pedidos = Usuario.query.options(
    joinedload(Usuario.pedidos)
).all()

for usuario in usuarios_con_pedidos:
    print(f"{usuario.nombre} tiene {len(usuario.pedidos)} pedidos")  # 1 consulta

Consultas con campos específicos:

# Seleccionar solo campos necesarios
nombres_emails = db.session.query(
    Usuario.nombre, 
    Usuario.email
).filter(Usuario.activo == True).all()

# Resultado como tuplas nombradas
for nombre, email in nombres_emails:
    print(f"{nombre}: {email}")

# Consulta con alias para campos calculados
usuarios_info = db.session.query(
    Usuario.nombre,
    Usuario.email,
    (Usuario.edad * 365).label('dias_vividos')
).all()

Consultas en lotes (batch processing):

# Procesar grandes volúmenes de datos en lotes
def procesar_usuarios_en_lotes(tamaño_lote=1000):
    offset = 0
    
    while True:
        usuarios = Usuario.query.offset(offset).limit(tamaño_lote).all()
        
        if not usuarios:
            break
            
        # Procesar lote actual
        for usuario in usuarios:
            # Lógica de procesamiento
            usuario.procesado = True
        
        db.session.commit()
        offset += tamaño_lote
        
        print(f"Procesados {offset} usuarios")

Consultas SQL nativas y flexibilidad

En ocasiones, las consultas SQL nativas proporcionan mayor control o mejor rendimiento para operaciones específicas que son difíciles de expresar con el ORM.

Ejecución de SQL directo:

from sqlalchemy import text

# Consulta SQL nativa con parámetros
resultado = db.session.execute(
    text("SELECT nombre, email FROM usuarios WHERE edad BETWEEN :min_edad AND :max_edad"),
    {"min_edad": 25, "max_edad": 35}
).fetchall()

for fila in resultado:
    print(f"{fila.nombre} - {fila.email}")

# Consulta compleja con joins nativos
consulta_compleja = text("""
    SELECT u.nombre, COUNT(p.id) as total_pedidos
    FROM usuarios u
    LEFT JOIN pedidos p ON u.id = p.usuario_id
    WHERE u.activo = :activo
    GROUP BY u.id, u.nombre
    HAVING COUNT(p.id) > :min_pedidos
    ORDER BY total_pedidos DESC
""")

resultados = db.session.execute(
    consulta_compleja,
    {"activo": True, "min_pedidos": 5}
).fetchall()

Combinación de ORM y SQL nativo:

# Usar resultados de SQL nativo para filtrar con ORM
ids_usuarios_activos = db.session.execute(
    text("SELECT id FROM usuarios WHERE ultimo_acceso > NOW() - INTERVAL 30 DAY")
).scalars().all()

usuarios_recientes = Usuario.query.filter(
    Usuario.id.in_(ids_usuarios_activos)
).all()

La comprensión profunda del sistema de consultas y sesiones de SQLAlchemy permite desarrollar aplicaciones Flask más eficientes y mantenibles. Estas herramientas avanzadas se vuelven especialmente valiosas cuando trabajamos con grandes volúmenes de datos o cuando necesitamos optimizar el rendimiento de nuestras aplicaciones web.