Concepto clave

Las operaciones CRUD avanzadas en FastAPI van más allá de las operaciones básicas de crear, leer, actualizar y eliminar. Cuando trabajas con sistemas de producción, necesitas manejar relaciones complejas entre entidades y garantizar la integridad de los datos mediante transacciones. Piensa en esto como una transferencia bancaria: no puedes simplemente restar dinero de una cuenta sin asegurarte de que se suma a otra. Si algo falla en medio, todo debe revertirse para evitar inconsistencias.

Las relaciones en bases de datos (como uno-a-muchos o muchos-a-muchos) requieren que diseñes tus modelos SQLAlchemy con cuidado y optimices las consultas para evitar el problema N+1. Imagina un sistema de e-commerce donde un pedido tiene múltiples productos. Si consultas cada producto por separado, el rendimiento se desploma. En cambio, debes usar joins y carga ansiosa para traer todo en una sola consulta.

Cómo funciona en la práctica

Vamos a construir un sistema de gestión de proyectos donde los usuarios pueden crear proyectos y asignarles tareas. Cada proyecto tiene un usuario propietario (relación uno-a-muchos) y múltiples tareas (otra relación uno-a-muchos). Implementaremos operaciones CRUD que respeten estas relaciones y usen transacciones para mantener la consistencia.

Paso 1: Define los modelos SQLAlchemy con relaciones. Usa relationship y ForeignKey para conectar las tablas. Paso 2: Crea esquemas Pydantic que incluyan datos anidados (como mostrar tareas dentro de un proyecto). Paso 3: Implementa endpoints en FastAPI que usen transacciones con db.session para operaciones atómicas. Por ejemplo, al crear un proyecto con tareas iniciales, todo debe guardarse o nada.

Código en acción

Modelos SQLAlchemy con relaciones:

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

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    projects = relationship("Project", back_populates="owner")

class Project(Base):
    __tablename__ = "projects"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))
    owner = relationship("User", back_populates="projects")
    tasks = relationship("Task", back_populates="project", cascade="all, delete-orphan")

class Task(Base):
    __tablename__ = "tasks"
    id = Column(Integer, primary_key=True, index=True)
    description = Column(String)
    project_id = Column(Integer, ForeignKey("projects.id"))
    project = relationship("Project", back_populates="tasks")

Endpoint con transacción para crear proyecto y tareas:

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from models import Project, Task
from schemas import ProjectCreate, ProjectResponse
import logging

router = APIRouter()

@router.post("/projects/", response_model=ProjectResponse)
def create_project_with_tasks(project_data: ProjectCreate, db: Session = Depends(get_db)):
    try:
        # Inicia transacción
        db_project = Project(title=project_data.title, owner_id=project_data.owner_id)
        db.add(db_project)
        db.flush()  # Obtiene ID para usar en tareas
        
        for task_desc in project_data.tasks:
            db_task = Task(description=task_desc, project_id=db_project.id)
            db.add(db_task)
        
        db.commit()  # Confirma todo
        db.refresh(db_project)
        return db_project
    except Exception as e:
        db.rollback()  # Revierte en caso de error
        logging.error(f"Error en transacción: {e}")
        raise HTTPException(status_code=500, detail="Error al crear proyecto")

Errores comunes

No usar transacciones en operaciones múltiples: Si creas un proyecto y sus tareas en inserciones separadas sin transacción, un fallo puede dejar datos inconsistentes. Siempre envuelve operaciones relacionadas en una transacción.
Problema N+1 en consultas: Al recuperar proyectos con sus tareas, hacer una consulta por cada tarea es ineficiente. Usa joinedload de SQLAlchemy para cargar todo en una consulta.
Olvidar manejo de excepciones: Sin un bloque try-except y rollback, los errores pueden dejar la transacción abierta, bloqueando recursos.
No validar relaciones en esquemas: Asegúrate de que los esquemas Pydantic reflejen las relaciones para validar datos anidados antes de llegar a la base de datos.
Ignorar concurrencia: En sistemas con múltiples usuarios, usa with_for_update o versionado optimista para evitar condiciones de carrera.

Checklist de dominio

¿Puedes definir modelos SQLAlchemy con relaciones uno-a-muchos y muchos-a-muchos correctamente?
¿Implementas transacciones para operaciones atómicas que involucran múltiples entidades?
¿Optimizas consultas con joins o carga ansiosa para evitar el problema N+1?
¿Manejas excepciones y rollbacks en cada operación transaccional?
¿Usas esquemas Pydantic anidados para validar y serializar datos relacionados?
¿Diseñas endpoints que devuelven datos anidados (como proyectos con tareas) eficientemente?
¿Pruebas escenarios de error, como fallos a mitad de transacción, en tu código?

Sistema de Gestión de Biblioteca con Préstamos y Devoluciones

Implementa un sistema CRUD avanzado para una biblioteca donde los libros pueden ser prestados a usuarios, manejando relaciones y transacciones.

Crea los modelos SQLAlchemy: Book (con campos como id, título, autor, disponible), User (id, nombre, email), y Loan (id, book_id, user_id, fecha_préstamo, fecha_devolución). Establece relaciones: un usuario puede tener múltiples préstamos, un libro puede estar en un préstamo a la vez.
Implementa un endpoint POST /loans/ que cree un préstamo. Debe: verificar que el libro esté disponible, crear el registro de préstamo, y actualizar el libro a no disponible. Usa una transacción para asegurar que ambas operaciones sean atómicas.
Crea un endpoint PUT /loans/{loan_id}/return para devolver un libro. Debe: marcar el préstamo como devuelto (agregar fecha_devolución), y actualizar el libro a disponible. Nuevamente, usa transacción.
Implementa un endpoint GET /users/{user_id}/loans que liste todos los préstamos de un usuario con detalles del libro, optimizado para evitar consultas N+1.
Prueba tu implementación con casos como préstamo de libro no disponible o errores durante la transacción.

Pistas

Usa ForeignKey en el modelo Loan para conectar con Book y User.
En la transacción, considera usar db.flush() para obtener IDs antes de confirmar.
Para optimizar la consulta de préstamos, investiga joinedload en SQLAlchemy.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.

Realizar Operaciones CRUD Avanzadas con Relaciones y Transacciones