Operaciones CRUD y consultas

Create, Read, Update, Delete básicos

Las operaciones CRUD (Create, Read, Update, Delete) son las cuatro acciones fundamentales que puedes realizar sobre los datos en tu base de datos. Con SQLAlchemy y FastAPI, estas operaciones se traducen en métodos específicos que te permiten manipular registros de forma sencilla y eficiente.

Operación Create (Crear)

La creación de registros implica instanciar un modelo SQLAlchemy, añadirlo a la sesión y confirmar los cambios. El proceso sigue un patrón consistente que puedes aplicar a cualquier modelo.

from sqlalchemy.orm import Session
from models import Usuario
from schemas import UsuarioCreate

def crear_usuario(db: Session, usuario_data: UsuarioCreate):
    # Crear instancia del modelo
    db_usuario = Usuario(
        nombre=usuario_data.nombre,
        email=usuario_data.email,
        edad=usuario_data.edad
    )
    
    # Añadir a la sesión
    db.add(db_usuario)
    
    # Confirmar cambios
    db.commit()
    
    # Refrescar para obtener el ID generado
    db.refresh(db_usuario)
    
    return db_usuario

El método db.add() registra el objeto en la sesión, mientras que db.commit() ejecuta la inserción en la base de datos. La función db.refresh() actualiza el objeto con los valores generados automáticamente, como el ID.

Operación Read (Leer)

La lectura de datos puede realizarse de múltiples formas según tus necesidades. SQLAlchemy proporciona métodos intuitivos para obtener registros individuales o múltiples.

Obtener un registro por ID:

def obtener_usuario(db: Session, usuario_id: int):
    return db.query(Usuario).filter(Usuario.id == usuario_id).first()

Obtener todos los registros:

def obtener_usuarios(db: Session, skip: int = 0, limit: int = 100):
    return db.query(Usuario).offset(skip).limit(limit).all()

El método query() inicia la consulta, filter() aplica condiciones, y first() o all() ejecutan la consulta devolviendo uno o múltiples resultados respectivamente. Los parámetros skip y limit implementan paginación básica.

Operación Update (Actualizar)

La actualización de registros requiere primero obtener el objeto existente, modificar sus atributos y confirmar los cambios. Este enfoque garantiza que trabajas con datos actuales.

def actualizar_usuario(db: Session, usuario_id: int, usuario_data: UsuarioCreate):
    # Obtener el registro existente
    db_usuario = db.query(Usuario).filter(Usuario.id == usuario_id).first()
    
    if db_usuario:
        # Actualizar atributos
        db_usuario.nombre = usuario_data.nombre
        db_usuario.email = usuario_data.email
        db_usuario.edad = usuario_data.edad
        
        # Confirmar cambios
        db.commit()
        db.refresh(db_usuario)
    
    return db_usuario

También puedes usar el método update() para actualizaciones masivas:

def actualizar_usuarios_por_edad(db: Session, edad_minima: int, nuevo_estado: str):
    db.query(Usuario).filter(Usuario.edad >= edad_minima).update(
        {"estado": nuevo_estado}
    )
    db.commit()

Operación Delete (Eliminar)

La eliminación de registros sigue un patrón similar a la actualización: obtener el objeto y usar el método delete() de la sesión.

def eliminar_usuario(db: Session, usuario_id: int):
    db_usuario = db.query(Usuario).filter(Usuario.id == usuario_id).first()
    
    if db_usuario:
        db.delete(db_usuario)
        db.commit()
        return True
    
    return False

Para eliminaciones masivas basadas en condiciones:

def eliminar_usuarios_inactivos(db: Session):
    usuarios_eliminados = db.query(Usuario).filter(
        Usuario.activo == False
    ).delete()
    db.commit()
    return usuarios_eliminados

Integración con endpoints FastAPI

Las funciones CRUD se integran naturalmente con tus endpoints FastAPI. Cada operación corresponde típicamente a un método HTTP específico.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db

app = FastAPI()

@app.post("/usuarios/", response_model=UsuarioResponse)
def crear_usuario_endpoint(usuario: UsuarioCreate, db: Session = Depends(get_db)):
    return crear_usuario(db=db, usuario_data=usuario)

@app.get("/usuarios/{usuario_id}", response_model=UsuarioResponse)
def obtener_usuario_endpoint(usuario_id: int, db: Session = Depends(get_db)):
    db_usuario = obtener_usuario(db=db, usuario_id=usuario_id)
    if db_usuario is None:
        raise HTTPException(status_code=404, detail="Usuario no encontrado")
    return db_usuario

@app.put("/usuarios/{usuario_id}", response_model=UsuarioResponse)
def actualizar_usuario_endpoint(
    usuario_id: int, 
    usuario: UsuarioCreate, 
    db: Session = Depends(get_db)
):
    db_usuario = actualizar_usuario(db=db, usuario_id=usuario_id, usuario_data=usuario)
    if db_usuario is None:
        raise HTTPException(status_code=404, detail="Usuario no encontrado")
    return db_usuario

@app.delete("/usuarios/{usuario_id}")
def eliminar_usuario_endpoint(usuario_id: int, db: Session = Depends(get_db)):
    eliminado = eliminar_usuario(db=db, usuario_id=usuario_id)
    if not eliminado:
        raise HTTPException(status_code=404, detail="Usuario no encontrado")
    return {"mensaje": "Usuario eliminado correctamente"}

Manejo de errores comunes

Las operaciones CRUD pueden fallar por diversas razones. Es importante manejar estos casos para proporcionar respuestas apropiadas.

from sqlalchemy.exc import IntegrityError

def crear_usuario_seguro(db: Session, usuario_data: UsuarioCreate):
    try:
        db_usuario = Usuario(**usuario_data.dict())
        db.add(db_usuario)
        db.commit()
        db.refresh(db_usuario)
        return db_usuario
    except IntegrityError:
        db.rollback()
        raise HTTPException(
            status_code=400, 
            detail="Error: el email ya existe"
        )

El método db.rollback() revierte los cambios cuando ocurre un error, manteniendo la consistencia de la base de datos. Esta práctica es esencial para aplicaciones robustas que manejan datos críticos.

Consultas simples con filter() y first()

Las consultas específicas te permiten buscar registros que cumplan condiciones particulares en lugar de obtener todos los datos. SQLAlchemy proporciona métodos intuitivos para filtrar información y recuperar exactamente lo que necesitas de tu base de datos.

El método filter()

El método filter() aplica condiciones a tus consultas usando operadores de comparación familiares. Puedes combinar múltiples filtros para crear búsquedas precisas y eficientes.

from sqlalchemy.orm import Session
from models import Usuario

def buscar_usuarios_por_edad(db: Session, edad_minima: int):
    return db.query(Usuario).filter(Usuario.edad >= edad_minima).all()

def buscar_usuario_por_email(db: Session, email: str):
    return db.query(Usuario).filter(Usuario.email == email).first()

Los operadores de comparación disponibles incluyen igualdad (==), desigualdad (!=), mayor que (>), menor que (<), y sus variantes con igual (>=, <=). Cada operador se traduce automáticamente a SQL válido.

Filtros con operadores de texto

Para búsquedas de texto, SQLAlchemy ofrece métodos específicos que facilitan la localización de contenido parcial o con patrones específicos.

def buscar_usuarios_por_nombre(db: Session, nombre_parcial: str):
    # Búsqueda que contiene el texto
    return db.query(Usuario).filter(
        Usuario.nombre.contains(nombre_parcial)
    ).all()

def buscar_usuarios_que_empiecen_con(db: Session, prefijo: str):
    # Búsqueda que empieza con el texto
    return db.query(Usuario).filter(
        Usuario.nombre.startswith(prefijo)
    ).all()

def buscar_usuarios_que_terminen_con(db: Session, sufijo: str):
    # Búsqueda que termina con el texto
    return db.query(Usuario).filter(
        Usuario.nombre.endswith(sufijo)
    ).all()

El método contains() busca coincidencias parciales, mientras que startswith() y endswith() localizan texto al inicio o final respectivamente. Estos métodos son insensibles a mayúsculas y minúsculas por defecto.

Combinando múltiples filtros

Puedes encadenar múltiples filtros para crear consultas más específicas. SQLAlchemy interpreta múltiples filtros como condiciones AND por defecto.

def buscar_usuarios_adultos_activos(db: Session):
    return db.query(Usuario).filter(
        Usuario.edad >= 18,
        Usuario.activo == True
    ).all()

def buscar_usuarios_por_rango_edad(db: Session, edad_min: int, edad_max: int):
    return db.query(Usuario).filter(
        Usuario.edad >= edad_min,
        Usuario.edad <= edad_max
    ).all()

Para condiciones OR, utiliza el operador | junto con paréntesis para agrupar las condiciones correctamente:

from sqlalchemy import or_

def buscar_usuarios_jovenes_o_seniors(db: Session):
    return db.query(Usuario).filter(
        or_(Usuario.edad <= 25, Usuario.edad >= 65)
    ).all()

El método first() vs all()

La diferencia entre first() y all() determina cuántos resultados obtienes de tu consulta. Esta elección afecta tanto el rendimiento como el tipo de dato devuelto.

def obtener_primer_usuario_activo(db: Session):
    # Devuelve un objeto Usuario o None
    return db.query(Usuario).filter(Usuario.activo == True).first()

def obtener_todos_usuarios_activos(db: Session):
    # Devuelve una lista de objetos Usuario
    return db.query(Usuario).filter(Usuario.activo == True).all()

El método first() detiene la búsqueda tan pronto como encuentra una coincidencia, lo que resulta más eficiente cuando solo necesitas un registro. Si no encuentra resultados, devuelve None en lugar de generar una excepción.

Consultas con ordenamiento

Combinar filtros con ordenamiento te permite obtener resultados organizados según criterios específicos. Esto es especialmente útil cuando trabajas con first() para obtener el "mejor" resultado.

def obtener_usuario_mas_joven(db: Session):
    return db.query(Usuario).filter(
        Usuario.activo == True
    ).order_by(Usuario.edad.asc()).first()

def obtener_usuario_mas_reciente(db: Session):
    return db.query(Usuario).order_by(
        Usuario.fecha_registro.desc()
    ).first()

Los métodos asc() y desc() controlan el orden ascendente o descendente respectivamente. Puedes ordenar por múltiples campos separándolos con comas.

Implementación en endpoints FastAPI

Las consultas filtradas se integran perfectamente con FastAPI usando parámetros de consulta opcionales que permiten búsquedas 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.filter(Usuario.nombre.contains(nombre))
    if edad_min:
        query = query.filter(Usuario.edad >= edad_min)
    if edad_max:
        query = query.filter(Usuario.edad <= edad_max)
    if activo is not None:
        query = query.filter(Usuario.activo == activo)
    
    return query.all()

Validación de resultados

Es importante validar los resultados de tus consultas, especialmente cuando usas first(), para manejar apropiadamente los casos donde no se encuentran coincidencias.

@app.get("/usuarios/email/{email}")
def obtener_usuario_por_email(email: str, db: Session = Depends(get_db)):
    usuario = db.query(Usuario).filter(Usuario.email == email).first()
    
    if not usuario:
        raise HTTPException(
            status_code=404, 
            detail=f"No se encontró usuario con email: {email}"
        )
    
    return usuario

Esta validación explícita garantiza que tu API responda consistentemente y proporcione mensajes de error claros cuando los datos solicitados no existen en la base de datos.