Cómo construir una API REST asíncrona con FastAPI, SQLAlchemy Async y Alembic paso a paso

Te llevo de cero a una API CRUD asíncrona en Python usando FastAPI, SQLAlchemy (async), Alembic para migraciones y Docker. Incluye estructura de carpetas, código completo y por qué cada pieza está así.

Qué vas a crear

• API REST con endpoints CRUD para un recurso 'Task' (tarea).
• Conexión asíncrona a PostgreSQL usando SQLAlchemy Async (asyncpg).
• Migraciones con Alembic configurado para uso async.
• Contenedores Docker y docker-compose para desarrollo.

Requisitos

• Python 3.10+
• Docker + docker-compose (opcional pero recomendado)
• PostgreSQL (local o en Docker)

Decisiones rápidas: por qué estas herramientas

• FastAPI: rutas async, validación automática Pydantic y rendimiento.
• SQLAlchemy Async: control fino de ORM con conexiones asíncronas.
• Alembic: migraciones automatizadas y reproducibles.
• Docker: entorno reproducible.

Estructura de carpetas

project/
  app/
    __init__.py
    main.py
    database.py
    models.py
    schemas.py
    crud.py
  alembic.ini
  alembic/
    env.py
    versions/
  Dockerfile
  docker-compose.yml
  requirements.txt

requirements.txt

fastapi
uvicorn[standard]
SQLAlchemy>=1.4
asyncpg
alembic
pydantic

1) database.py — conexión asíncrona y sesión

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import declarative_base
import os

DATABASE_URL = os.getenv('DATABASE_URL', 'postgresql+asyncpg://postgres:password@db:5432/appdb')

engine = create_async_engine(DATABASE_URL, echo=True, future=True)
AsyncSessionLocal = async_sessionmaker(bind=engine, expire_on_commit=False)

Base = declarative_base()

async def get_session():
    async with AsyncSessionLocal() as session:
        yield session

Por qué: usamos create_async_engine con asyncpg y un async_sessionmaker para inyectar sesiones asíncronas en endpoints.

2) models.py — definición ORM

from sqlalchemy import Column, Integer, String, Boolean
from .database import Base

class Task(Base):
    __tablename__ = 'tasks'

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    description = Column(String(1000), nullable=True)
    completed = Column(Boolean, default=False, nullable=False)

3) schemas.py — Pydantic para request/response

from pydantic import BaseModel
from typing import Optional

class TaskBase(BaseModel):
    title: str
    description: Optional[str] = None

class TaskCreate(TaskBase):
    pass

class TaskUpdate(BaseModel):
    title: Optional[str]
    description: Optional[str]
    completed: Optional[bool]

class TaskOut(TaskBase):
    id: int
    completed: bool

    class Config:
        orm_mode = True

4) crud.py — operaciones con la DB (async)

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from . import models, schemas

async def get_task(session: AsyncSession, task_id: int):
    result = await session.execute(select(models.Task).where(models.Task.id == task_id))
    return result.scalars().first()

async def get_tasks(session: AsyncSession, skip: int = 0, limit: int = 100):
    result = await session.execute(select(models.Task).offset(skip).limit(limit))
    return result.scalars().all()

async def create_task(session: AsyncSession, task_in: schemas.TaskCreate):
    task = models.Task(**task_in.dict())
    session.add(task)
    await session.commit()
    await session.refresh(task)
    return task

async def update_task(session: AsyncSession, task: models.Task, task_in: schemas.TaskUpdate):
    for key, value in task_in.dict(exclude_unset=True).items():
        setattr(task, key, value)
    session.add(task)
    await session.commit()
    await session.refresh(task)
    return task

async def delete_task(session: AsyncSession, task: models.Task):
    await session.delete(task)
    await session.commit()

5) main.py — FastAPI con dependencias

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List

from . import models, schemas, crud
from .database import get_session, engine, Base

app = FastAPI(title='Tasks API')

# Crear tablas (solo para desarrollo rápido; en producción usa Alembic)
@app.on_event('startup')
async def on_startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.post('/tasks/', response_model=schemas.TaskOut)
async def create_task(task_in: schemas.TaskCreate, session: AsyncSession = Depends(get_session)):
    return await crud.create_task(session, task_in)

@app.get('/tasks/', response_model=List[schemas.TaskOut])
async def list_tasks(skip: int = 0, limit: int = 100, session: AsyncSession = Depends(get_session)):
    return await crud.get_tasks(session, skip, limit)

@app.get('/tasks/{task_id}', response_model=schemas.TaskOut)
async def get_task(task_id: int, session: AsyncSession = Depends(get_session)):
    task = await crud.get_task(session, task_id)
    if not task:
        raise HTTPException(status_code=404, detail='Task not found')
    return task

@app.patch('/tasks/{task_id}', response_model=schemas.TaskOut)
async def patch_task(task_id: int, task_in: schemas.TaskUpdate, session: AsyncSession = Depends(get_session)):
    task = await crud.get_task(session, task_id)
    if not task:
        raise HTTPException(status_code=404, detail='Task not found')
    return await crud.update_task(session, task, task_in)

@app.delete('/tasks/{task_id}', status_code=204)
async def delete_task(task_id: int, session: AsyncSession = Depends(get_session)):
    task = await crud.get_task(session, task_id)
    if not task:
        raise HTTPException(status_code=404, detail='Task not found')
    await crud.delete_task(session, task)

Por qué: endpoints asíncronos para no bloquear el event loop; las sesiones se inyectan con dependencias.

6) Alembic: configuración para SQLAlchemy Async

alembic.ini: mantén defaults; asegúrate de que tu sqlalchemy.url no sea obligatorio porque lo cargamos desde env.

alembic/env.py (esencial para async)

from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy import engine_from_config
from sqlalchemy import create_engine

from alembic import context
import os
import asyncio

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
fileConfig(config.config_file_name)

# add your model's MetaData object here for 'autogenerate' support
# from myapp import models
from app.database import Base
target_metadata = Base.metadata

DATABASE_URL = os.getenv('DATABASE_URL', 'postgresql+asyncpg://postgres:password@db:5432/appdb')

# Use async engine; run migrations in sync mode via connection.run_sync
def run_migrations_online():
    from sqlalchemy.ext.asyncio import create_async_engine

    connectable = create_async_engine(DATABASE_URL, poolclass=pool.NullPool)

    async def do_run_migrations():
        async with connectable.connect() as connection:
            await connection.run_sync(do_run_migrations_sync)
        await connectable.dispose()

    def do_run_migrations_sync(connection: Connection):
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

    asyncio.run(do_run_migrations())

if context.is_offline_mode():
    raise RuntimeError('Offline mode not supported for async setup')
else:
    run_migrations_online()

Con esto Alembic puede autogenerar migraciones y ejecutarlas conectándose con asyncpg.

7) Dockerfile y docker-compose (desarrollo)

Dockerfile

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY ./app ./app
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker-compose.yml

version: '3.8'
services:
  db:
    image: postgres:15
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: password
      POSTGRES_DB: appdb
    ports:
      - '5432:5432'
    volumes:
      - db_data:/var/lib/postgresql/data

  api:
    build: .
    depends_on:
      - db
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:password@db:5432/appdb
    ports:
      - '8000:8000'

volumes:
  db_data:

8) Migraciones básicas

1. Inicializa Alembic: alembic init alembic (si no lo hiciste)
2. Genera migración: alembic revision --autogenerate -m 'create tasks'
3. Aplica migración: alembic upgrade head

Nota: Alembic ejecutará env.py que usa el URL de entorno para crear la conexión async y aplicar las migraciones.

9) Probar la API

Levanta con Docker Compose: docker-compose up --build. La API estará en http://localhost:8000. Explora la documentación automática en /docs.

Buenas prácticas y advertencias

• No uses Base.metadata.create_all en producción: usa Alembic para migraciones confiables.
• Configura timeouts y tamaño de pool en el engine async según carga: create_async_engine(..., pool_size=20, max_overflow=10) (ajusta según app).
• Evita operaciones sincrónicas pesadas en endpoints async: delega a workers (Celery, RQ o background tasks).
• Siempre valida y sanitiza entradas con Pydantic y límites de paginación para evitar DoS.

Extensiones recomendadas

• Usar pydantic-settings o environs para gestión de configuración segura.
• Integrar tests (pytest + httpx AsyncClient) para endpoints async.
• Instrumentar con observabilidad (OpenTelemetry, Prometheus).

Consejo avanzado: en producción, considera usar un pool de conexiones ligero (pgbouncer en transaction pooling) delante de PostgreSQL para evitar agotamiento de conexiones por procesos async/uvicorn múltiples. También configura índices en columnas usadas por filtros y análisis de EXPLAIN para queries pesadas.

Advertencia: la programación asíncrona mejora la concurrencia IO-bound pero requiere que todas las piezas (drivers, ORMs, librerías) sean compatibles con async; mezclar sync en código async puede llevar a bloqueos sutiles.

Siguiente paso sugerido: añade tests de integración con una base de datos temporal (pytest + pytest-asyncio + testcontainers) o configura CI que aplique migraciones y ejecute pruebas antes de deployment.