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

Objetivo: crear una API CRUD asíncrona que use FastAPI, SQLAlchemy en modo async y PostgreSQL. Incluiré estructura de carpetas, código completo de los módulos clave, configuración de Alembic para migraciones y un test asíncrono básico. Explico el porqué de las decisiones y los puntos críticos.

Stack y por qué

• Python 3.10+ — tipos y soporte async mejorado.
• FastAPI — rendimiento y productividad (pydantic + OpenAPI).
• SQLAlchemy 1.4+ en modo async — ORM moderno con soporte tanto sync como async.
• asyncpg — driver async para PostgreSQL.
• Alembic — migraciones (adaptado al modo async).
• pytest + httpx (AsyncClient) — tests asíncronos.

Requisitos previos

• PostgreSQL corriendo localmente o en un contenedor.
• Python 3.10+ instalado.
• Instalar dependencias:

  python -m pip install fastapi uvicorn[standard] sqlalchemy[asyncio] asyncpg alembic pydantic pytest httpx pytest-asyncio

Estructura de proyecto

project/
├── app/
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── schemas.py
│   ├── crud.py
│   └── deps.py
├── alembic/
│   ├── env.py
│   └── versions/
├── alembic.ini
├── tests/
│   └── test_items.py
└── pyproject.toml / requirements.txt

Paso 1 — Configurar conexión y sesión async

Archivo: app/db.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

DATABASE_URL = 'postgresql+asyncpg://postgres:password@localhost:5432/mydb'

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    echo=False,
)

AsyncSessionLocal = sessionmaker(
    bind=engine,
    expire_on_commit=False,
    class_=AsyncSession,
)

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

Por qué: usar sessionmaker con class_=AsyncSession y expire_on_commit=False para evitar recargar objetos tras commit en entorno async.

Paso 2 — Modelos (ORM)

Archivo: app/models.py

from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Item(Base):
    __tablename__ = 'items'

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

Paso 3 — Schemas Pydantic

Archivo: app/schemas.py

from pydantic import BaseModel
from typing import Optional

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

class ItemRead(BaseModel):
    id: int
    title: str
    description: Optional[str] = None
    completed: bool

    class Config:
        orm_mode = True

class ItemUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    completed: Optional[bool] = None

Paso 4 — Operaciones CRUD

Archivo: app/crud.py

from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from . import models, schemas

async def get_item(session: AsyncSession, item_id: int):
    result = await session.execute(select(models.Item).where(models.Item.id == item_id))
    return result.scalars().first()

async def get_items(session: AsyncSession, skip: int = 0, limit: int = 100) -> List[models.Item]:
    result = await session.execute(select(models.Item).offset(skip).limit(limit))
    return result.scalars().all()

async def create_item(session: AsyncSession, item: schemas.ItemCreate) -> models.Item:
    db_item = models.Item(title=item.title, description=item.description)
    session.add(db_item)
    await session.commit()
    await session.refresh(db_item)
    return db_item

async def update_item(session: AsyncSession, item_id: int, item: schemas.ItemUpdate):
    db_item = await get_item(session, item_id)
    if not db_item:
        return None
    for field, value in item.dict(exclude_unset=True).items():
        setattr(db_item, field, value)
    session.add(db_item)
    await session.commit()
    await session.refresh(db_item)
    return db_item

async def delete_item(session: AsyncSession, item_id: int) -> bool:
    db_item = await get_item(session, item_id)
    if not db_item:
        return False
    await session.delete(db_item)
    await session.commit()
    return True

Paso 5 — Dependencias y endpoints

Archivo: app/deps.py

from .db import get_session

# aquí podrías agregar más dependencias como current_user, config, etc.

Archivo: app/main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from . import crud, schemas
from .db import get_session

app = FastAPI(title='Items API')

@app.get('/items', response_model=List[schemas.ItemRead])
async def list_items(skip: int = 0, limit: int = 100, session: AsyncSession = Depends(get_session)):
    return await crud.get_items(session, skip, limit)

@app.get('/items/{item_id}', response_model=schemas.ItemRead)
async def read_item(item_id: int, session: AsyncSession = Depends(get_session)):
    db_item = await crud.get_item(session, item_id)
    if not db_item:
        raise HTTPException(status_code=404, detail='Item not found')
    return db_item

@app.post('/items', response_model=schemas.ItemRead, status_code=201)
async def create_item(item: schemas.ItemCreate, session: AsyncSession = Depends(get_session)):
    return await crud.create_item(session, item)

@app.put('/items/{item_id}', response_model=schemas.ItemRead)
async def update_item(item_id: int, item: schemas.ItemUpdate, session: AsyncSession = Depends(get_session)):
    updated = await crud.update_item(session, item_id, item)
    if not updated:
        raise HTTPException(status_code=404, detail='Item not found')
    return updated

@app.delete('/items/{item_id}', status_code=204)
async def delete_item(item_id: int, session: AsyncSession = Depends(get_session)):
    ok = await crud.delete_item(session, item_id)
    if not ok:
        raise HTTPException(status_code=404, detail='Item not found')
    return None

Paso 6 — Alembic (migraciones) en modo async

Config clave: en alembic/env.py debes usar run_migrations_online adaptado para async. Un ejemplo mínimo:

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

import os
import sys
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from app.models import Base
from app.db import DATABASE_URL

config = context.config
fileConfig(config.config_file_name)

target_metadata = Base.metadata

def run_migrations_offline():
    raise NotImplementedError('Offline migrations are not supported for async setups')

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

def run_migrations_online():
    connectable = create_async_engine(DATABASE_URL)
    with connectable.connect() as connection:
        connection = connection.execution_options(isolation_level='AUTOCOMMIT')
        import asyncio
        asyncio.get_event_loop().run_until_complete(do_run_migrations(connection))

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

Nota: Alembic no es totalmente async; la receta típica es crear un engine async pero ejecutar migrations sincronamente mediante un loop. Ajusta según la versión y tu entorno.

Paso 7 — Tests básicos asíncronos

Archivo: tests/test_items.py

import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_create_and_get_item():
    async with AsyncClient(app=app, base_url='http://test') as client:
        resp = await client.post('/items', json={'title': 'Test', 'description': 'Desc'})
        assert resp.status_code == 201
        data = resp.json()
        item_id = data['id']

        resp2 = await client.get(f'/items/{item_id}')
        assert resp2.status_code == 200
        assert resp2.json()['title'] == 'Test'

Para tests de integración con DB usa una base temporal o fixtures que creen y eliminen esquemas, o una transacción que se haga rollback.

Ejecutar la aplicación

• Levantar la app:

  uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

• Ejecutar migraciones:

  alembic revision --autogenerate -m 'create items'
  alembic upgrade head
  

Buenas prácticas y puntos críticos

• Usa la URL con dialecto async: postgresql+asyncpg://user:pass@host/db.
• Controla pool_size y max_overflow según tu carga. Valores pequeños pueden estrangular, valores grandes también consumen conexiones DB.
• No mezcles sesiones sync y async en el mismo proceso sin cuidado.
• En endpoints CPU-bound evita bloquear el loop; usa background tasks o procesos separados.
• Evita consultas N+1: usa selectinload/joinedload según convenga.
• Para pruebas, considera usar una base SQLite in-memory solo para lógica que no dependa de características específicas de Postgres.

Despliegue y rendimiento

Usa un servidor ASGI (uvicorn, gunicorn + uvicorn workers) y configura workers en función de la CPU. Para alta concurrencia prioriza conexiones DB y tamaño de pool: cada worker mantiene su propio pool, así que dimensiona pool_size en consecuencia.

Consejo avanzado: si tienes muchas conexiones concurrentes y operaciones rápidas en la DB, considera un pool reducido y reutilización cuidadosa de sesiones; para operaciones largas, usa colas o workers separados para no agotar el pool.

Advertencia: Alembic y SQLAlchemy async tienen matices; revisa la versión de SQLAlchemy y la receta oficial de Alembic para async antes de llevarlo a producción.

Siguiente paso sugerido: añade autenticación, paginación y tests de integración con contenedores (docker-compose) para simular el entorno real.