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

Este tutorial muestra un proyecto mínimo pero completo: FastAPI + SQLAlchemy en modo asíncrono (SQLite para demo). Verás estructura de carpetas, código completo y explicaciones del por qué de las decisiones.

Por qué asíncrono

• Mejor uso de I/O: permite atender muchas peticiones concurrentes cuando la mayor parte del trabajo es I/O (DB, HTTP, archivos).
• Requiere drivers/ORM compatibles con async (ej. asyncpg para Postgres, aiosqlite para SQLite).
• Evita bloquear el loop: no ejecutes tareas CPU-bound en el loop principal.

Estructura de carpetas

project/
  app/
    __init__.py
    main.py
    database.py
    models.py
    schemas.py
    crud.py
  requirements.txt

requirements.txt

fastapi
uvicorn[standard]
sqlalchemy>=1.4
aiosqlite
pydantic

(Para Postgres reemplaza aiosqlite por asyncpg y la URL de la DB).

1) database.py

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

DATABASE_URL = 'sqlite+aiosqlite:///./test.db'

engine = create_async_engine(DATABASE_URL, echo=True, future=True)

# sessionmaker que genera AsyncSession
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Base = declarative_base()

async def init_db():
    # Crea las tablas; en producción usa Alembic
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

Por qué: usamos expire_on_commit=False para evitar que los objetos se expiren después de commit y así poder leer atributos sin reconsultar la BD. La función init_db es útil para demos; para migraciones reales usa Alembic en modo async.

2) models.py

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

class Item(Base):
    __tablename__ = 'items'
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(100), index=True, nullable=False)
    description = Column(Text, nullable=True)

3) schemas.py (Pydantic)

from pydantic import BaseModel
from typing import Optional

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

class ItemRead(ItemCreate):
    id: int

    class Config:
        orm_mode = True

4) crud.py (operaciones DB)

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

async def get_item(db: AsyncSession, item_id: int) -> Optional[Item]:
    result = await db.execute(select(Item).where(Item.id == item_id))
    return result.scalars().first()

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

async def create_item(db: AsyncSession, title: str, description: str | None = None) -> Item:
    item = Item(title=title, description=description)
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

async def update_item(db: AsyncSession, item: Item, title: str, description: str | None = None) -> Item:
    item.title = title
    item.description = description
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

async def delete_item(db: AsyncSession, item: Item) -> None:
    await db.delete(item)
    await db.commit()

Por qué: usamos patrones simples (CRUD) y refresh para asegurar datos actualizados tras commit. Alternativamente puedes usar sentencias update/delete para operaciones en bloque.

5) main.py (FastAPI)

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

from . import models, schemas, crud, database

app = FastAPI(title='Items API (async)')

@app.on_event('startup')
async def on_startup():
    await database.init_db()

# dependency: sesion por request
async def get_db() -> AsyncSession:
    async with database.AsyncSessionLocal() as session:
        yield session

@app.post('/items/', response_model=schemas.ItemRead, status_code=status.HTTP_201_CREATED)
async def create_item_endpoint(item: schemas.ItemCreate, db: AsyncSession = Depends(get_db)):
    created = await crud.create_item(db, title=item.title, description=item.description)
    return created

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

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

@app.put('/items/{item_id}', response_model=schemas.ItemRead)
async def update_item_endpoint(item_id: int, payload: schemas.ItemCreate, db: AsyncSession = Depends(get_db)):
    item = await crud.get_item(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail='Item not found')
    updated = await crud.update_item(db, item, title=payload.title, description=payload.description)
    return updated

@app.delete('/items/{item_id}', status_code=status.HTTP_204_NO_CONTENT)
async def delete_item_endpoint(item_id: int, db: AsyncSession = Depends(get_db)):
    item = await crud.get_item(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail='Item not found')
    await crud.delete_item(db, item)
    return None

Ejecutar la app

# Instala dependencias
pip install -r requirements.txt

# Ejecuta con uvicorn
uvicorn app.main:app --reload

Prueba con curl o HTTP client:

curl -X POST "http://127.0.0.1:8000/items/" -H 'Content-Type: application/json' -d '{"title":"Hola","description":"Descripción"}'
curl "http://127.0.0.1:8000/items/"

Notas, mejores prácticas y advertencias

• Drivers: para producción con Postgres usa 'postgresql+asyncpg://user:pass@host/db'.
• Migraciones: usa Alembic con configuración async para mantener historial de esquemas.
• Sesión por petición: no compartas la misma sesión entre requests; usa dependency que crea/ciere sesión.
• Concurrency: uvicorn + uvloop manejan bien concurrencia I/O. Para CPU-heavy usa procesos o background tasks.
• Pooling: create_async_engine maneja pool; configura parámetro pool_size / max_overflow según tu DB.
• Transacciones: usa session.begin() para operaciones compuestas y rollback automático en excepciones.

Ejemplo rápido de transacción compuesta:

async with db.begin():
    db.add(obj1)
    db.add(obj2)
# commit automático al salir del with si no hubo excepciones

Siguientes pasos recomendados

• Agregar autenticación (OAuth2 / JWT) y control de permisos en endpoints.
• Configurar Alembic para migraciones y CI/CD para despliegues (tests, linters, pre-commit).
• Escribir tests asíncronos con pytest + httpx.AsyncClient.

Consejo avanzado: para cargas altas en I/O, evita aumentar procesos; prioriza tuning del pool y del driver async (ej. asyncpg pool params). Advertencia: no ejecutes operaciones CPU-bound en endpoints sin moverlas a workers (Celery, RQ, background tasks) o fuera del event loop.

Si quieres, te doy la versión lista para producción (Alembic, Dockerfile, ejemplo con Postgres y tests) o escribo los tests asíncronos paso a paso.