Cómo Construir APIs Rápidas y Escalables con Python y FastAPI

FastAPI se convirtió en el framework de Python más popular para APIs — y por buenas razones. Es 3-5x más rápido que Flask, genera documentación automática, y tiene validación de datos integrada con Pydantic. En esta guía construimos una API completa lista para producción.

Por qué FastAPI y no Flask o Django REST

Característica	FastAPI	Flask	Django REST
Performance	⚡ Async nativo	Sync	Sync
Validación	Pydantic automática	Manual	Serializers
Documentación	OpenAPI automática	Manual/Swagger	Manual
Typing	Nativo	Opcional	Parcial
Curva de aprendizaje	Baja	Baja	Media-Alta

Flask sigue siendo excelente para proyectos simples. Django REST para proyectos con ORM complejo y admin incluido. Pero para APIs modernas de alto rendimiento, FastAPI es la mejor opción en 2026.

Setup inicial

# Crear proyecto
mkdir mi-api && cd mi-api
python -m venv venv
source venv/bin/activate

# Instalar dependencias
pip install fastapi uvicorn pydantic-settings sqlalchemy asyncpg

Estructura del proyecto

mi-api/
├── app/
│   ├── __init__.py
│   ├── main.py          # Entry point
│   ├── config.py         # Configuración
│   ├── models/           # SQLAlchemy models
│   ├── schemas/          # Pydantic schemas
│   ├── routes/           # Endpoints
│   ├── services/         # Lógica de negocio
│   └── middleware/       # Auth, CORS, etc.
├── tests/
├── Dockerfile
├── docker-compose.yml
└── requirements.txt

Modelo + Schema + Endpoint

Modelo SQLAlchemy:

# app/models/product.py
from sqlalchemy import Column, Integer, String, Float, DateTime
from sqlalchemy.sql import func
from app.database import Base

class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(255), nullable=False)
    price = Column(Float, nullable=False)
    category = Column(String(100))
    created_at = Column(DateTime, server_default=func.now())

Schema Pydantic (validación automática):

# app/schemas/product.py
from pydantic import BaseModel, Field
from datetime import datetime

class ProductCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=255)
    price: float = Field(..., gt=0)
    category: str | None = None

class ProductResponse(ProductCreate):
    id: int
    created_at: datetime

    model_config = {"from_attributes": True}

Endpoint:

# app/routes/products.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.schemas.product import ProductCreate, ProductResponse
from app.services.product_service import ProductService
from app.database import get_db

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

@router.post("/", response_model=ProductResponse, status_code=201)
async def create_product(
    product: ProductCreate,
    db: AsyncSession = Depends(get_db)
):
    service = ProductService(db)
    return await service.create(product)

@router.get("/", response_model=list[ProductResponse])
async def list_products(
    skip: int = 0,
    limit: int = 20,
    category: str | None = None,
    db: AsyncSession = Depends(get_db)
):
    service = ProductService(db)
    return await service.list(skip=skip, limit=limit, category=category)

@router.get("/{product_id}", response_model=ProductResponse)
async def get_product(
    product_id: int,
    db: AsyncSession = Depends(get_db)
):
    service = ProductService(db)
    product = await service.get_by_id(product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

Autenticación con JWT

# app/middleware/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from app.config import settings

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    try:
        payload = jwt.decode(
            credentials.credentials,
            settings.jwt_secret,
            algorithms=["HS256"]
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token expired"
        )
    except jwt.InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )

Middleware esencial

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routes import products, auth

app = FastAPI(
    title="Mi API",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://midominio.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(products.router, prefix="/api/v1")
app.include_router(auth.router, prefix="/api/v1")

@app.get("/health")
async def health_check():
    return {"status": "ok"}

Testing con pytest

# tests/test_products.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.mark.asyncio
async def test_create_product():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post("/api/v1/products/", json={
            "name": "Laptop Pro",
            "price": 999.99,
            "category": "electronics"
        })
        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "Laptop Pro"
        assert data["price"] == 999.99

Dockerfile para producción

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Performance tips

1. Usa async/await: FastAPI soporta endpoints asíncronos nativamente. Usa asyncpg para PostgreSQL y aiohttp para HTTP calls.

2. Connection pooling: Configura pools de conexiones a la base de datos.

3. Response caching: Usa Redis para cachear respuestas frecuentes.

4. Background tasks: Usa BackgroundTasks para operaciones que no necesitan respuesta inmediata.

from fastapi import BackgroundTasks

@router.post("/orders/")
async def create_order(order: OrderCreate, bg: BackgroundTasks):
    result = await order_service.create(order)
    bg.add_task(send_confirmation_email, result.email)
    return result

5. Paginación eficiente: Siempre limita los resultados con skip y limit.

De FastAPI a producción

FastAPI + Docker + Cloud Run es una combinación que escala de 0 a miles de requests por segundo sin gestionar infraestructura manualmente. Si combinas esto con un pipeline de CI/CD, tienes una ruta clara de commit a producción en minutos.

¿Necesitas construir una API robusta para tu producto? En CloudLabs diseñamos y construimos APIs que escalan. Hablemos.