FastAPI guía

# Instalar FastAPI y servidor ASGI
pip install fastapi uvicorn

# Con extras para producción
pip install fastapi[all]

from fastapi import FastAPI

app = FastAPI(
    title="Data Service",
    version="1.0.0",
)

@app.get("/")
def root():
    return {"service": "data-api", "status": "ok"}

FastAPI genera docs interactivos sin configuración extra

En producción, es recomendable deshabilitar /docs o restringir el acceso a ambientes internos.

data-api/
├── app/
│   ├── __init__.py
│   ├── main.py          # App FastAPI + routers
│   ├── config.py         # Settings con Pydantic
│   ├── routers/
│   │   ├── health.py     # /health para Kong
│   │   ├── ingest.py     # Endpoints de ingesta
│   │   └── query.py      # Endpoints de consulta
│   ├── models/
│   │   ├── schemas.py    # Pydantic models (request/response)
│   │   └── database.py   # SQLAlchemy / conexiones
│   ├── services/
│   │   └── data_service.py
│   └── dependencies.py   # Auth, DB sessions
├── tests/
├── Dockerfile
├── requirements.txt
└── .env

@app.get("/datos")
def listar_datos():
    return {"datos": []}

@app.post("/datos")
def crear_dato(dato: dict):
    return {"creado": True}

@app.put("/datos/{id}")
def actualizar_dato(id: int, dato: dict):
    return {"actualizado": id}

@app.delete("/datos/{id}")
def eliminar_dato(id: int):
    return {"eliminado": id}

@app.get("/clientes/{cliente_id}")
def obtener_cliente(cliente_id: int):
    """FastAPI valida el tipo automáticamente"""
    return {"cliente_id": cliente_id}

# Enum para valores fijos
from enum import Enum

class Ambiente(str, Enum):
    dev = "dev"
    staging = "staging"
    prod = "prod"

@app.get("/config/{ambiente}")
def config(ambiente: Ambiente):
    return {"ambiente": ambiente.value}

from typing import Optional

@app.get("/buscar")
def buscar(
    q: str,                          # Obligatorio
    limite: int = 10,                # Default 10
    offset: int = 0,                 # Default 0
    activo: Optional[bool] = None,   # Opcional
):
    return {
        "query": q,
        "limite": limite,
        "offset": offset,
        "activo": activo,
    }

Ejemplo: GET /buscar?q=fibra&limite=20&activo=true

from fastapi import Header, Cookie

@app.get("/info")
def info(
    x_trace_id: str = Header(None),
    x_consumer: str = Header(None),
    session_id: str = Cookie(None),
):
    return {
        "trace_id": x_trace_id,
        "consumer": x_consumer,
        "session": session_id,
    }

Un API gateway puede inyectar headers como X-Consumer y X-Trace-Id en cada request. FastAPI los convierte automáticamente (guiones a guiones bajos).

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

class ClienteRequest(BaseModel):
    nombre: str = Field(..., min_length=1, max_length=100)
    documento: str = Field(..., pattern=r"^\d{7,13}$")
    email: Optional[str] = None
    activo: bool = True

class ClienteResponse(BaseModel):
    id: int
    nombre: str
    documento: str
    creado_en: datetime

    class Config:
        from_attributes = True  # Pydantic v2

@app.post(
    "/clientes",
    response_model=ClienteResponse,
    status_code=201,
    summary="Crear cliente",
    tags=["clientes"],
)
def crear_cliente(req: ClienteRequest):
    # FastAPI valida el body automáticamente
    # Si falla, retorna 422 con detalle de errores
    return {
        "id": 1,
        "nombre": req.nombre,
        "documento": req.documento,
        "creado_en": datetime.now(),
    }

from fastapi import HTTPException
from fastapi.responses import JSONResponse

@app.get("/clientes/{id}")
def obtener_cliente(id: int):
    cliente = buscar_en_db(id)
    if not cliente:
        raise HTTPException(
            status_code=404,
            detail={"error": "Cliente no encontrado", "id": id},
        )
    return cliente

# Handler global de excepciones
@app.exception_handler(Exception)
async def error_global(request, exc):
    return JSONResponse(
        status_code=500,
        content={
            "error": "Error interno",
            "trace_id": request.headers.get("x-trace-id"),
        },
    )

from fastapi import Depends, HTTPException

# Dependencia: verificar API key de Kong
def verificar_api_key(
    x_api_key: str = Header(None),
):
    if not x_api_key:
        raise HTTPException(401, "API key requerida")
    if x_api_key not in KEYS_VALIDAS:
        raise HTTPException(403, "API key inválida")
    return x_api_key

@app.get("/datos-protegidos")
def datos(api_key: str = Depends(verificar_api_key)):
    return {"datos": [...], "autenticado_con": api_key}

from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/clientes")
def listar_clientes(db: Session = Depends(get_db)):
    return db.query(Cliente).all()

import time
import logging

logger = logging.getLogger("data-api")

@app.middleware("http")
async def log_requests(request, call_next):
    start = time.time()
    trace_id = request.headers.get("x-trace-id", "sin-trace")

    response = await call_next(request)

    duration = time.time() - start
    logger.info(
        f"[{trace_id}] {request.method} {request.url.path} "
        f"→ {response.status_code} ({duration:.3f}s)"
    )
    return response

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://your-frontend.example.com",
        "<tu-url>",
    ],
    allow_methods=["*"],
    allow_headers=["*"],
)

# routers/health.py
from fastapi import APIRouter

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

@router.get("")
def healthcheck():
    return {"status": "healthy", "service": "data-api"}

# main.py
from app.routers import health, ingest, query

app.include_router(health.router)
app.include_router(ingest.router, prefix="/api/v1")
app.include_router(query.router, prefix="/api/v1")

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "data-api"
    debug: bool = False
    database_url: str
    kong_api_key: str
    log_level: str = "INFO"

    class Config:
        env_file = ".env"

settings = Settings()

Kong requiere un health check para su load balancer

from datetime import datetime

@app.get("/health")
def health():
    return {
        "status": "healthy",
        "timestamp": datetime.utcnow().isoformat(),
        "version": settings.app_name,
    }

Configurar en Kong: health_checks.active.http_path = "/health"

from typing import Generic, TypeVar, Optional
from pydantic import BaseModel

T = TypeVar("T")

class ApiResponse(BaseModel, Generic[T]):
    success: bool
    data: Optional[T] = None
    error: Optional[str] = None
    trace_id: Optional[str] = None

# Uso
@app.get("/clientes", response_model=ApiResponse[list[dict]])
def listar_clientes(request: Request):
    return ApiResponse(
        success=True,
        data=[{"id": 1, "nombre": "Empresa X"}],
        trace_id=request.headers.get("x-trace-id"),
    )

from fastapi import BackgroundTasks

def procesar_archivo(path: str):
    """Tarea pesada que corre en background"""
    # Procesar CSV, notificar, etc.
    pass

@app.post("/ingest/csv")
def ingest_csv(
    path: str,
    background_tasks: BackgroundTasks,
):
    background_tasks.add_task(procesar_archivo, path)
    return {"status": "procesando", "path": path}

FROM python:3.11-slim

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

COPY app/ ./app/

EXPOSE <puerto>
CMD ["uvicorn", "app.main:app", "--host", "<host>", "--port", "<puerto>", "--workers", "4"]

El Dockerfile debe exponer el puerto configurado. Un reverse proxy o API gateway se encarga de enrutar el tráfico externo.