Cómo construir una API REST asíncrona con FastAPI, SQLAlchemy async y PostgreSQL

En este tutorial montarás una API REST asíncrona completa con FastAPI, SQLAlchemy en modo async y PostgreSQL. Verás la estructura del proyecto, el código completo de los componentes clave (modelos, CRUD, rutas), y cómo gestionar migraciones con Alembic usando el motor async. Al final podrás ejecutar la API y probar rutas básicas.

Por qué usar async aquí

• Permite manejar muchas conexiones concurrentes (I/O-bound) sin bloquear el event loop.
• Ideal para APIs que hacen muchas consultas a DB u otras llamadas de red.
• FastAPI ya está diseñado para async y combina bien con SQLAlchemy 1.4+ en modo async.

Requisitos

• Python 3.9+
• PostgreSQL (local o remoto)
• pip, virtualenv

Instalación rápida

python -m venv venv
source venv/bin/activate
pip install fastapi uvicorn[standard] sqlalchemy[asyncio] asyncpg alembic pydantic python-dotenv

Estructura del proyecto

backend/
├─ app/
│  ├─ __init__.py
│  ├─ main.py
│  ├─ core/
│  │  ├─ config.py
│  ├─ db/
│  │  ├─ base.py
│  │  ├─ session.py
│  ├─ models/
│  │  ├─ user.py
│  ├─ schemas/
│  │  ├─ user.py
│  ├─ crud/
│  │  ├─ user.py
│  ├─ api/
│  │  ├─ deps.py
│  │  ├─ v1/
│  │  │  ├─ user.py
├─ alembic/
│  ├─ env.py
│  ├─ versions/
├─ .env
├─ requirements.txt

Variables de entorno (.env)

DATABASE_URL=postgresql+asyncpg://dbuser:dbpass@localhost:5432/mydb

config.py (lectura de .env)

from pydantic import BaseSettings

class Settings(BaseSettings):
    database_url: str

    class Config:
        env_file = ".env"

settings = Settings()

session.py (motor y session async)

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

engine = create_async_engine(settings.database_url, echo=False, future=True)
AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_session() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

base.py (Base declarativa)

from sqlalchemy.orm import declarative_base

Base = declarative_base()

models/user.py

from sqlalchemy import Column, Integer, String
from app.db.base import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    full_name = Column(String, nullable=True)

schemas/user.py (Pydantic)

from pydantic import BaseModel, EmailStr
from typing import Optional

class UserBase(BaseModel):
    email: EmailStr
    full_name: Optional[str] = None

class UserCreate(UserBase):
    pass

class UserRead(UserBase):
    id: int

    class Config:
        orm_mode = True

crud/user.py (operaciones DB)

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.schemas.user import UserCreate

async def get_user_by_id(db: AsyncSession, user_id: int) -> User | None:
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalars().first()

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

async def create_user(db: AsyncSession, user_in: UserCreate) -> User:
    user = User(**user_in.dict())
    db.add(user)
    try:
        await db.commit()
        await db.refresh(user)
    except IntegrityError:
        await db.rollback()
        raise
    return user

api/deps.py

from fastapi import Depends
from app.db.session import get_session
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncSession:
    async for s in get_session():
        yield s

api/v1/user.py (rutas)

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.schemas.user import UserCreate, UserRead
from app.crud.user import create_user, get_user_by_email, get_user_by_id
from app.api.deps import get_db

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user_endpoint(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
    existing = await get_user_by_email(db, user_in.email)
    if existing:
        raise HTTPException(status_code=400, detail="Email already registered")
    user = await create_user(db, user_in)
    return user

@router.get("/{user_id}", response_model=UserRead)
async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await get_user_by_id(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

main.py

from fastapi import FastAPI
from app.api.v1 import user as user_router

app = FastAPI(title="FastAPI Async SQLAlchemy Example")

app.include_router(user_router.router)

# Evento opcional para iniciar recursos al arrancar
@app.on_event("startup")
async def startup_event():
    # Aquí podrías crear conexiones, caches, etc.
    pass

Migrations con Alembic (async)

Configura alembic para usar un AsyncEngine y env.py que detecte modelos declarativos.

alembic/env.py (fragmento clave)

from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy import engine_from_config
from sqlalchemy.ext.asyncio import AsyncEngine
from alembic import context

# 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)

from app.db.base import Base
from app.core.config import settings

target_metadata = Base.metadata

def run_migrations_offline():
    url = settings.database_url
    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)

    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    connectable = AsyncEngine(
        engine_from_config(
            config.get_section(config.config_ini_section),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
        )
    )

    with connectable.connect() as connection:  # noqa: E501
        context.configure(connection=connection, target_metadata=target_metadata)

        with context.begin_transaction():
            context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    import asyncio

    asyncio.run(run_migrations_online())

Nota: Alembic por defecto no crea AsyncEngine; en varias guías se usa una pequeña adaptación para crear AsyncEngine con create_async_engine y luego run migrations con connection.get_sync_connection() o usando API async. Ajusta según versión de Alembic.

Comandos útiles

# Inicializar alembic (si no lo hiciste)
alembic init alembic

# Crear una revisión (autogenerate busca metadata)
alembic revision --autogenerate -m "create users"

# Aplicar migraciones
alembic upgrade head

# Ejecutar la app
uvicorn app.main:app --reload

Cómo probar (curl)

curl -X POST "http://127.0.0.1:8000/users/" -H "Content-Type: application/json" -d '{"email":"alice@example.com","full_name":"Alice"}'

curl "http://127.0.0.1:8000/users/1"

Razones técnicas y decisiones clave

• expire_on_commit=False: evita que los objetos se “expiren” al commit — útil en ASGI donde reusar datos después del commit es común.
• sessionmaker con class_=AsyncSession: crea sesiones compatibles con await/async.
• Separación CRUD/routers/schemas: facilita testing y mantenimiento.
• Usar asyncpg + SQLAlchemy async: performant y simple; evita usar capas extra si necesitas control fino de SQLAlchemy.

Problemas comunes y cómo evitarlos

• Bloquear el event loop: no llames funciones síncronas pesadas dentro de endpoints async (usa run_in_executor).
• Transacciones olvidadas: maneja commit/rollback y usa try/except en operaciones críticas.
• Migrations que no detectan modelos: asegúrate de importar todos los módulos que definen modelos en env.py o en tu package para que Base.metadata contenga todas las tablas.

Siguientes pasos recomendados

• Añadir paginación y filtros con SQLAlchemy.
• Integrar pruebas async (pytest + pytest-asyncio) para endpoints y CRUD.
• Configurar CI que ejecute migraciones y pruebas contra una DB temporal (Postgres en Docker).

Consejo avanzado: monitoriza y ajusta el tamaño del pool de conexiones (connection pool) según la concurrencia esperada y el número de workers de uvicorn (si usas varios procesos cada uno tendrá su pool). Evita pools muy grandes que sobrecarguen PostgreSQL y tampoco pools demasiado pequeños que provoquen latencia por espera de conexiones.

Advertencia: si alguna operación usa librerías síncronas que hacen I/O (por ejemplo, algunos ORMs o librerías de terceros), encapsúralas con run_in_executor o muévelas a un worker para no bloquear el event loop.