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.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.