Concepto clave

Conectar FastAPI a bases de datos SQL utilizando SQLAlchemy con soporte asíncrono es fundamental para construir APIs escalables que manejen múltiples solicitudes concurrentes sin bloquear el servidor. Imagina un restaurante donde los meseros (FastAPI) pueden atender a varios clientes (solicitudes HTTP) simultáneamente mientras los cocineros (la base de datos) preparan los pedidos en paralelo, en lugar de hacer fila uno tras otro. Esto es posible gracias al modelo asíncrono de Python, que permite que tu aplicación siga procesando otras tareas mientras espera respuestas de la base de datos.

SQLAlchemy es el ORM (Mapeo Objeto-Relacional) más popular en Python, que actúa como un traductor entre los objetos de tu código Python y las tablas de la base de datos. Con su extensión asíncrona, puedes ejecutar consultas sin bloquear el hilo principal, lo que es crucial para aplicaciones de alto rendimiento. La combinación de FastAPI (que es asíncrono por diseño) con SQLAlchemy asíncrono crea una arquitectura donde cada componente opera de manera eficiente, similar a cómo un equipo de cirugía coordina múltiples especialistas trabajando al mismo tiempo en diferentes aspectos de una operación.

Cómo funciona en la práctica

Para implementar esta conexión, seguimos una arquitectura modular que separa claramente la configuración de la base de datos, los modelos de datos y las operaciones CRUD. Primero, configuramos el motor de base de datos usando create_async_engine de SQLAlchemy, especificando la URL de conexión (por ejemplo, para PostgreSQL con asyncpg). Luego, creamos una sesión asíncrona con async_sessionmaker que manejará las transacciones.

Los modelos se definen usando la sintaxis declarativa de SQLAlchemy, mapeando clases Python a tablas SQL. Para las operaciones, utilizamos el patrón de repositorio, donde cada función (como crear, leer, actualizar o eliminar registros) se implementa como una corrutina asíncrona. Esto permite que FastAPI llame a estas funciones con await, liberando el ciclo de eventos para otras tareas mientras se completa la consulta a la base de datos. Un flujo típico: el cliente hace una solicitud POST a tu API, FastAPI valida los datos, llama a una función asíncrona que inserta un nuevo registro en la base de datos usando la sesión, y devuelve la respuesta una vez que la operación se completa, todo sin bloquear otras solicitudes entrantes.

Codigo en accion

Configuración básica de la base de datos y un modelo de ejemplo:

# database.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# Dependencia para obtener sesión en rutas FastAPI
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

Modelo de usuario y operación CRUD asíncrona:

# models.py
from sqlalchemy import String, Integer
from sqlalchemy.orm import Mapped, mapped_column
from database import Base

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True)
    name: Mapped[str] = mapped_column(String)

# crud.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from models import User

async def create_user(db: AsyncSession, email: str, name: str) -> User:
    user = User(email=email, name=name)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    result = await db.execute(select(User).where(User.email == email))
    return result.scalar_one_or_none()

Errores comunes

1. Olvidar await en operaciones de base de datos: En código asíncrono, todas las llamadas a la base de datos (como commit, execute, o add) deben precederse de await. Sin esto, la operación no se ejecutará correctamente y puede causar timeouts o errores silenciosos. Solución: Revisa que cada llamada a métodos de sesión use await.

2. No manejar adecuadamente las sesiones: Crear sesiones sin cerrarlas puede llevar a fugas de conexiones. Usa el patrón async with para asegurar que las sesiones se cierren automáticamente, o implementa la dependencia get_db como se muestra en el código.

3. Confundir SQLAlchemy síncrono con asíncrono: Importar desde sqlalchemy en lugar de sqlalchemy.ext.asyncio resultará en errores de compatibilidad. Asegúrate de usar AsyncSession, create_async_engine, y otros componentes asíncronos específicos.

4. No configurar correctamente el pool de conexiones: Por defecto, SQLAlchemy usa un pool que puede no ser óptimo para cargas altas. Ajusta parámetros como pool_size y max_overflow en create_async_engine según tu caso de uso.

5. Ignorar el manejo de transacciones: En operaciones complejas que involucran múltiples consultas, no usar transacciones explícitas puede dejar la base de datos en estado inconsistente. Usa async with db.begin(): para agrupar operaciones relacionadas.

Checklist de dominio

Puedo configurar una conexión asíncrona a PostgreSQL o MySQL usando SQLAlchemy con una URL correcta
Sé definir modelos SQLAlchemy con tipos de datos mapeados y relaciones básicas
Implemento operaciones CRUD completas (crear, leer, actualizar, eliminar) como funciones asíncronas
Uso la dependencia get_db en rutas FastAPI para inyectar sesiones de base de datos
Manejo errores de base de datos (como IntegrityError) con try-except en las operaciones
Configuro el pool de conexiones ajustando parámetros para rendimiento en producción
Escribo pruebas unitarias asíncronas para las operaciones de base de datos usando pytest con fixtures

Implementar un sistema de gestión de productos con operaciones asíncronas

En este ejercicio, crearás una API REST para gestionar productos en una tienda online, utilizando FastAPI con SQLAlchemy asíncrono. Sigue estos pasos:

Configura un proyecto FastAPI con una base de datos PostgreSQL local o en contenedor Docker. Usa la URL postgresql+asyncpg://user:password@localhost/product_db.
Define un modelo Product con los campos: id (entero, clave primaria), name (string, único), description (string opcional), price (float), y stock (entero).
Crea funciones CRUD asíncronas en un módulo crud.py para: crear producto, obtener producto por ID, listar todos los productos con paginación (usando limit y offset), actualizar stock, y eliminar producto.
Implementa rutas FastAPI en main.py que usen estas funciones, con validación de datos usando Pydantic. Incluye endpoints para POST /products, GET /products, GET /products/{id}, PUT /products/{id}/stock, y DELETE /products/{id}.
Asegúrate de manejar errores comunes, como producto no encontrado (devolver 404) o violación de unicidad (devolver 409).
Prueba la API con herramientas como curl o Postman, verificando que las operaciones sean atómicas y no bloqueantes.

Pistas

Usa async with en la dependencia get_db para manejar la sesión automáticamente
Para la paginación, implementa parámetros de consulta skip y limit en el endpoint GET /products
Considera usar transacciones explícitas al actualizar el stock para evitar condiciones de carrera

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.

Conectar FastAPI a Bases de Datos SQL con SQLAlchemy y Async