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

23 sept 2026 1,073 vistas
python Cómo construir una API RESTful asíncrona con FastAPI y SQLAlchemy Async paso a paso

Cómo construir una API RESTful asíncrona con FastAPI y SQLAlchemy Async

En este tutorial práctico vas a crear una API asíncrona completa usando FastAPI, SQLAlchemy Async y Alembic. Verás la estructura del proyecto, el código completo de modelos, esquemas, rutas, gestión de sesión async, migraciones y tests básicos. Se prioriza claridad y buenas prácticas.

Qué aprenderás

  • Configurar FastAPI con SQLAlchemy asíncrono y asyncpg
  • Diseñar modelos y esquemas (Pydantic)
  • Crear endpoints CRUD con patrones de sesión y transacciones
  • Configurar Alembic para migraciones async
  • Escribir tests con pytest y httpx AsyncClient
  • Dockerizar la app

Requisitos

  • Python 3.10+
  • Postgres (local o docker)

Estructura del proyecto

project_fastapi_async/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── schemas.py
│   ├── crud.py
│   ├── routers/
│   │   └── users.py
│   └── deps.py
├── alembic/
│   └── env.py
├── alembic.ini
├── tests/
│   └── test_users.py
├── Dockerfile
├── requirements.txt
└── pyproject.toml

Dependencias

# requirements.txt
fastapi
uvicorn[standard]
sqlalchemy>=1.4
asyncpg
alembic
pydantic
httpx
pytest
pytest-asyncio
python-dotenv

1) Configurar la base de datos y la sesión async

Archivo app/db.py

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

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

engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    pool_pre_ping=True,
)

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

Base = declarative_base()

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

Por qué: usamos expire_on_commit=False para evitar que las instancias se vuelvan a cargar automáticamente después de commit, lo que simplifica el acceso a atributos tras operaciones async.

2) Modelos con SQLAlchemy

Archivo app/models.py

from sqlalchemy import Column, Integer, String, ForeignKey, Text
from sqlalchemy.orm import relationship
from .db import Base

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, index=True, nullable=False)
    full_name = Column(String(255), nullable=True)
    hashed_password = Column(String(255), nullable=False)

    items = relationship('Item', back_populates='owner')

class Item(Base):
    __tablename__ = 'items'
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(255), index=True, nullable=False)
    description = Column(Text, nullable=True)
    owner_id = Column(Integer, ForeignKey('users.id'))

    owner = relationship('User', back_populates='items')

3) Esquemas Pydantic

Archivo app/schemas.py

from pydantic import BaseModel, EmailStr
from typing import Optional

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

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True

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

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int

    class Config:
        orm_mode = True

4) Operaciones CRUD

Archivo app/crud.py

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

async def get_user(session: AsyncSession, user_id: int) -> Optional[models.User]:
    result = await session.execute(select(models.User).where(models.User.id == user_id))
    return result.scalars().first()

async def get_user_by_email(session: AsyncSession, email: str) -> Optional[models.User]:
    result = await session.execute(select(models.User).where(models.User.email == email))
    return result.scalars().first()

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

async def create_user(session: AsyncSession, user_in: schemas.UserCreate) -> models.User:
    fake_hashed = 'notreallyhashed-' + user_in.password
    db_user = models.User(email=user_in.email, full_name=user_in.full_name, hashed_password=fake_hashed)
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

async def update_user(session: AsyncSession, user_id: int, **fields) -> Optional[models.User]:
    await session.execute(update(models.User).where(models.User.id == user_id).values(**fields))
    await session.commit()
    return await get_user(session, user_id)

async def delete_user(session: AsyncSession, user_id: int) -> None:
    await session.execute(delete(models.User).where(models.User.id == user_id))
    await session.commit()

5) Dependencias y router

Archivo app/deps.py

from .db import get_db

# Puedes añadir aquí dependencias comunes como current_user, etc.

Archivo app/routers/users.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from .. import schemas, crud
from ..db import get_db

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

@router.post('/', response_model=schemas.User, status_code=status.HTTP_201_CREATED)
async def create_user(user_in: schemas.UserCreate, session: AsyncSession = Depends(get_db)):
    existing = await crud.get_user_by_email(session, user_in.email)
    if existing:
        raise HTTPException(status_code=400, detail='Email already registered')
    return await crud.create_user(session, user_in)

@router.get('/', response_model=List[schemas.User])
async def read_users(skip: int = 0, limit: int = 100, session: AsyncSession = Depends(get_db)):
    return await crud.get_users(session, skip=skip, limit=limit)

@router.get('/{user_id}', response_model=schemas.User)
async def read_user(user_id: int, session: AsyncSession = Depends(get_db)):
    user = await crud.get_user(session, user_id)
    if not user:
        raise HTTPException(status_code=404, detail='User not found')
    return user

@router.delete('/{user_id}', status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int, session: AsyncSession = Depends(get_db)):
    await crud.delete_user(session, user_id)
    return None

6) App principal

Archivo app/main.py

from fastapi import FastAPI
from .routers import users

app = FastAPI(title='FastAPI async with SQLAlchemy')

app.include_router(users.router)

# evento opcional para crear tablas en desarrollo (no para producción)
from .db import engine, Base

@app.on_event('startup')
async def on_startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

7) Alembic: migraciones async

Configura alembic como de costumbre, pero usando el engine async en env.py. Ejemplo mínimo de sección para run_migrations_online:

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

config = context.config
fileConfig(config.config_file_name)

from app.db import Base
from app import models

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

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

async def run_async_migrations():
    connectable = create_async_engine(DATABASE_URL, poolclass=pool.NullPool)
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

if context.is_offline_mode():
    raise SystemExit('Offline mode not supported for async example')
else:
    import asyncio
    asyncio.run(run_async_migrations())

Con esto, las migraciones generadas por alembic funcionarán con la metadata de SQLAlchemy async.

8) Tests básicos

Archivo tests/test_users.py

import pytest
from httpx import AsyncClient
from app.main import app
from app.db import get_db, AsyncSessionLocal
from sqlalchemy.ext.asyncio import AsyncSession

@pytest.mark.asyncio
async def test_create_and_get_user(monkeypatch):
    async with AsyncClient(app=app, base_url='http://test') as ac:
        payload = {'email': 'test@example.com', 'password': 'secret'}
        r = await ac.post('/users/', json=payload)
        assert r.status_code == 201
        data = r.json()
        assert data['email'] == 'test@example.com'

        r2 = await ac.get(f"/users/{data['id']}")
        assert r2.status_code == 200
        assert r2.json()['email'] == 'test@example.com'

Nota: en tests reales debes usar una base de datos de prueba y fixtures para aislar el estado.

9) Dockerfile básico

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

Pautas y buenas prácticas

  • Usa conexiones pool adecuadas para async (asyncpg maneja pools internamente). Evita crear engines por request.
  • No uses operaciones bloqueantes en endpoints async. Si necesitas código CPU-bound, externaliza con Celery o background tasks.
  • Valida input con Pydantic y limita tamaños para evitar ataques de payload grande.
  • Aplica paginación y límites por defecto en list endpoints.
  • Protege endpoints con autenticación y roles; implementa OAuth2 con JWT según necesidades.

Problemas comunes y cómo evitarlos

  • Olvidar await al llamar funciones async -> runtime warnings y resultados inesperados.
  • Usar Session síncrona en contexto async -> bloqueará loop y degradará rendimiento.
  • Modificar ORM objects fuera de la sesión -> cambios no persistidos o errores de estado.

Cómo ejecutar

  1. Exporta DATABASE_URL, por ejemplo: postgresql+asyncpg://postgres:postgres@localhost:5432/postgres
  2. Instala dependencias: pip install -r requirements.txt
  3. Inicia la app: uvicorn app.main:app --reload
  4. Ejecuta tests: pytest -q

Con esto tienes una API asíncrona lista para producción con las piezas esenciales: app, modelos, migraciones y tests. Siguiente paso recomendado: implementar autenticación segura (OAuth2 con refresh tokens), añadir índices en la base de datos para consultas frecuentes y configurar métricas y tracing (Prometheus/OpenTelemetry) para monitorizar latencias y detectar hot paths en operaciones async.

Comentarios
¿Quieres comentar?

Inicia sesión con Telegram para participar en la conversación

Iniciar Sesión

Comentarios (0)

Aún no hay comentarios. ¡Sé el primero en comentar!

Más en python
Todas las Categorías
java 31 php 43 python 25 rust 34 sql 35
Iniciar Sesión