Quiz: Bases de Datos y Operaciones en FastAPI

Tipo de lección: Quiz

Concepto clave

En el desarrollo backend con FastAPI, la integración con bases de datos no es solo una conexión técnica, sino un puente entre la lógica de negocio y la persistencia de datos. Imagina una biblioteca donde los libros (datos) están organizados en estanterías (tablas), y el bibliotecario (FastAPI) necesita un sistema eficiente para encontrar, prestar y devolver libros sin causar caos. Las operaciones avanzadas en este contexto incluyen transacciones, consultas complejas y optimización de rendimiento, que son esenciales para APIs escalables en entornos de producción.

Para un nivel advanced, dominar esto significa entender cómo FastAPI interactúa con ORMs como SQLAlchemy o motores asíncronos, manejando concurrencia y evitando cuellos de botella. Una analogía real es un restaurante de alta cocina: el chef (FastAPI) coordina múltiples platos (peticiones) usando ingredientes frescos (datos) de la despensa (base de datos), asegurando que cada pedido se sirva rápido y sin errores, incluso en horas pico.

Cómo funciona en la práctica

En la práctica, integrar FastAPI con una base de datos implica configurar una conexión, definir modelos de datos y ejecutar operaciones CRUD con enfoque en seguridad y escalabilidad. Para un profesional, esto se traduce en pasos concretos: primero, establecer una sesión de base de datos usando herramientas como SQLAlchemy con soporte asíncrono; segundo, diseñar esquemas Pydantic para validación; tercero, implementar endpoints que manejen consultas eficientes, como joins o agregaciones.

Por ejemplo, en un sistema de e-commerce, podrías tener un endpoint para obtener órdenes de compra con detalles de productos. El proceso paso a paso incluye: 1) Configurar el motor de base de datos en main.py; 2) Crear modelos SQLAlchemy para Order y Product; 3) Usar async with para manejar sesiones de forma asíncrona; 4) Escribir consultas que optimicen el rendimiento, evitando N+1 queries con joinedload; 5) Validar respuestas con Pydantic para asegurar integridad.

Codigo en accion

Aquí tienes un ejemplo funcional que muestra una integración avanzada con PostgreSQL usando SQLAlchemy asíncrono y FastAPI, enfocado en una operación de consulta con joins:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base, joinedload
from sqlalchemy import Column, Integer, String, ForeignKey
from pydantic import BaseModel
from typing import List
import asyncio

# Configuración de la base de datos
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()

# Modelos SQLAlchemy
class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    price = Column(Integer, nullable=False)

class Order(Base):
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True, index=True)
    product_id = Column(Integer, ForeignKey("products.id"), nullable=False)
    quantity = Column(Integer, nullable=False)
    product = relationship("Product", lazy="joined")

# Esquemas Pydantic
class ProductSchema(BaseModel):
    id: int
    name: str
    price: int
    class Config:
        orm_mode = True

class OrderSchema(BaseModel):
    id: int
    product: ProductSchema
    quantity: int
    class Config:
        orm_mode = True

# Dependencia para sesión de base de datos
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

app = FastAPI()

# Endpoint avanzado con join y optimización
@app.get("/orders/{order_id}", response_model=OrderSchema)
async def get_order(order_id: int, db: AsyncSession = Depends(get_db)):
    query = select(Order).options(joinedload(Order.product)).where(Order.id == order_id)
    result = await db.execute(query)
    order = result.scalar_one_or_none()
    if not order:
        raise HTTPException(status_code=404, detail="Order not found")
    return order

Antes de la optimización, el código podría hacer consultas separadas para cada producto, causando el problema N+1. Después, con joinedload, se carga todo en una sola consulta, mejorando el rendimiento significativamente.

Errores comunes

• No usar sesiones asíncronas en operaciones I/O: En APIs avanzadas, bloquear el hilo con sesiones síncronas reduce la escalabilidad. Solución: Siempre usa AsyncSession de SQLAlchemy con motores como asyncpg.
• Ignorar el manejo de transacciones: Operaciones múltiples sin transacciones pueden dejar la base de datos en estado inconsistente. Solución: Envuelve operaciones críticas en async with db.begin() para rollback automático en errores.
• Sobrecargar endpoints con lógica de base de datos: Mezclar lógica de negocio con consultas directas hace el código difícil de mantener. Solución: Separa en capas, usando repositorios o servicios.
• No indexar columnas frecuentemente consultadas: En tablas grandes, falta de índices ralentiza las consultas. Solución: Añade índices en Column con index=True para campos como claves foráneas.
• Validación insuficiente en esquemas Pydantic: Confiar solo en la base de datos para validación puede llevar a errores de seguridad. Solución: Usa validadores personalizados en Pydantic para reglas de negocio.

Checklist de dominio

1. Configurar una conexión asíncrona a PostgreSQL o MySQL usando SQLAlchemy y FastAPI.
2. Implementar modelos SQLAlchemy con relaciones y claves foráneas correctamente definidas.
3. Crear endpoints que ejecuten consultas complejas con joins, evitando problemas N+1.
4. Manejar transacciones para operaciones atómicas, asegurando consistencia de datos.
5. Optimizar consultas con índices y técnicas de paginación para grandes volúmenes de datos.
6. Validar entradas y salidas usando esquemas Pydantic con configuraciones avanzadas.
7. Probar la integración con bases de datos usando pruebas asíncronas y fixtures.

Implementa un Sistema de Reservas con FastAPI y PostgreSQL

En este ejercicio práctico, construirás un sistema de reservas para un hotel, aplicando conceptos avanzados de integración con bases de datos en FastAPI. Sigue estos pasos:

1. Configura el entorno: Crea un proyecto FastAPI y configura una conexión asíncrona a PostgreSQL usando SQLAlchemy. Define la URL de base de datos en un archivo .env.
2. Diseña los modelos: Crea modelos SQLAlchemy para Room (con campos: id, type, price) y Booking (con campos: id, room_id, guest_name, check_in, check_out). Establece una relación entre ellos.
3. Implementa endpoints: Desarrolla endpoints para:
   • POST /bookings/: Crear una nueva reserva, validando que la habitación esté disponible en las fechas solicitadas.
   • GET /bookings/{booking_id}: Obtener detalles de una reserva incluyendo información de la habitación, usando una consulta con join optimizada.
   • GET /rooms/available: Listar habitaciones disponibles en un rango de fechas, con paginación para manejar muchos resultados.
4. Añade transacciones: En el endpoint de creación de reserva, usa una transacción para asegurar que la actualización de disponibilidad sea atómica.
5. Optimiza y prueba: Añade índices a los campos relevantes y escribe pruebas asíncronas para verificar la funcionalidad.

Entrega el código completo en un repositorio Git, con un README que explique las decisiones técnicas.

Pistas

• Usa SQLAlchemy's `select` con `where` y `not exists` para verificar disponibilidad de habitaciones.
• Para paginación, implementa parámetros `skip` y `limit` en el endpoint, y úsalos en la consulta con `offset` y `limit`.
• Considera usar `async with db.begin()` para manejar transacciones de forma explícita en operaciones críticas.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.