Realizar Operaciones CRUD Avanzadas con Relaciones y Transacciones

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

1. ¿Puedes definir modelos SQLAlchemy con relaciones uno-a-muchos y muchos-a-muchos correctamente?
2. ¿Implementas transacciones para operaciones atómicas que involucran múltiples entidades?
3. ¿Optimizas consultas con joins o carga ansiosa para evitar el problema N+1?
4. ¿Manejas excepciones y rollbacks en cada operación transaccional?
5. ¿Usas esquemas Pydantic anidados para validar y serializar datos relacionados?
6. ¿Diseñas endpoints que devuelven datos anidados (como proyectos con tareas) eficientemente?
7. ¿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.

1. 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.
2. 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.
3. 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.
4. 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.
5. Prueba tu implementación con casos como préstamo de libro no disponible o errores durante la transacción.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.