APIs REST

📐 Fundamento

Restricciones REST (Roy Fielding, 2000):

1. Cliente-servidor: separación clara de responsabilidades.
2. Stateless: cada request contiene toda la información necesaria; el servidor no guarda estado de sesión.
3. Cacheable: las respuestas deben indicar si pueden cachearse.
4. Sistema en capas: el cliente no sabe si habla con el servidor original o un proxy.
5. Uniform interface: los recursos se identifican con URIs, se manipulan con verbos HTTP estándar.
6. HATEOAS (opcional pero parte del estándar): respuestas incluyen links a acciones siguientes.

Mapeo de operaciones a HTTP:

• Seguro: no modifica datos.
• Idempotente: ejecutar N veces = ejecutar 1 vez.

Códigos de respuesta HTTP:

Convenciones de URLs:

GOOD                                BAD
/pedidos                            /getPedidos        ← verbos en URL
/pedidos/123                        /pedidos?id=123    ← ID como query
/pedidos/123/items                  /pedidoItems/123   ← jerarquía perdida
/pedidos?estado=ABIERTO&page=2      /pedidosAbiertosPagina2  ← acciones en URL

Recursos en plural y minúscula con guiones:
/menu-de-platillos                  ✓
/menuDePlatillos                    ✗
/MenuDePlatillos                    ✗

📐 Fundamento

Setup mínimo:

pip install fastapi uvicorn[standard] sqlalchemy psycopg2 pydantic[email]

API completa de pedidos:

from fastapi import FastAPI, HTTPException, Depends, status
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime
from enum import Enum

app = FastAPI(
    title="La Esquina API",
    version="1.0.0",
    description="API de pedidos para La Esquina"
)

# === MODELOS PYDANTIC (validación automática) ===

class EstadoPedido(str, Enum):
    ABIERTO = "ABIERTO"
    LISTO = "LISTO"
    ENTREGADO = "ENTREGADO"
    CANCELADO = "CANCELADO"

class ItemPedido(BaseModel):
    platillo_id: int = Field(..., gt=0, description="ID del platillo")
    cantidad: int = Field(..., gt=0, le=20, description="1 a 20 unidades")

class PedidoCrear(BaseModel):
    mesa_id: int = Field(..., gt=0)
    items: list[ItemPedido] = Field(..., min_length=1)
    notas: Optional[str] = Field(None, max_length=500)

class PedidoRespuesta(BaseModel):
    id: int
    mesa_id: int
    estado: EstadoPedido
    total: float
    creado_en: datetime
    items: list[ItemPedido]
    
    class Config:
        from_attributes = True  # permite construir desde ORM

# === ENDPOINTS ===

@app.post("/pedidos", response_model=PedidoRespuesta, status_code=201)
def crear_pedido(pedido: PedidoCrear, user=Depends(get_current_user)):
    """Crear un nuevo pedido."""
    nuevo = db.crear_pedido(
        mesa_id=pedido.mesa_id,
        items=pedido.items,
        notas=pedido.notas,
        mozo_id=user.id
    )
    return nuevo

@app.get("/pedidos", response_model=list[PedidoRespuesta])
def listar_pedidos(
    estado: Optional[EstadoPedido] = None,
    page: int = 1,
    size: int = 20,
    user=Depends(get_current_user)
):
    """Listar pedidos con filtros y paginación."""
    if size > 100:
        raise HTTPException(400, "Máximo 100 por página")
    return db.listar_pedidos(estado=estado, offset=(page-1)*size, limit=size)

@app.get("/pedidos/{pedido_id}", response_model=PedidoRespuesta)
def obtener_pedido(pedido_id: int, user=Depends(get_current_user)):
    pedido = db.get_pedido(pedido_id)
    if not pedido:
        raise HTTPException(404, "Pedido no encontrado")
    if pedido.mozo_id != user.id and not user.is_admin:
        raise HTTPException(403, "Sin permisos")
    return pedido

@app.patch("/pedidos/{pedido_id}", response_model=PedidoRespuesta)
def actualizar_estado(
    pedido_id: int,
    nuevo_estado: EstadoPedido,
    user=Depends(get_current_user)
):
    pedido = db.get_pedido(pedido_id)
    if not pedido:
        raise HTTPException(404)
    
    # Validar transición de estado
    if not estado_transicion_valida(pedido.estado, nuevo_estado):
        raise HTTPException(422, f"No se puede pasar de {pedido.estado} a {nuevo_estado}")
    
    return db.actualizar_estado(pedido_id, nuevo_estado)

@app.delete("/pedidos/{pedido_id}", status_code=204)
def cancelar_pedido(pedido_id: int, user=Depends(get_current_user)):
    db.cancelar_pedido(pedido_id, user.id)
    return None

# Iniciar el servidor:
# uvicorn main:app --reload --host 0.0.0.0 --port 8000

Lo que FastAPI te da gratis:

• Validación automática con Pydantic — pedido.items siempre será una lista válida.
• Documentación automática en /docs (Swagger UI) y /redoc.
• Schema OpenAPI en /openapi.json — usable para generar clientes en cualquier lenguaje.
• Async nativo — soporta async def para máxima concurrencia.
• Type hints — el editor te ayuda con autocomplete y detección de errores.

📐 Fundamento

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from passlib.context import CryptContext
from datetime import datetime, timedelta

pwd_context = CryptContext(schemes=["bcrypt"])
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
SECRET_KEY = os.environ["SECRET_KEY"]

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verificar_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def crear_access_token(data: dict, expires_minutes: int = 15) -> str:
    payload = {**data, "exp": datetime.utcnow() + timedelta(minutes=expires_minutes)}
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = db.get_user_by_email(form.username)
    if not user or not verificar_password(form.password, user.hashed_password):
        raise HTTPException(401, "Credenciales inválidas",
                          headers={"WWW-Authenticate": "Bearer"})
    
    access_token = crear_access_token({"sub": str(user.id), "role": user.role})
    return {"access_token": access_token, "token_type": "bearer"}

def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(401, "Token inválido")
    except JWTError:
        raise HTTPException(401, "Token inválido")
    
    user = db.get_user(int(user_id))
    if not user:
        raise HTTPException(401, "Usuario no encontrado")
    return user

# Cualquier endpoint con `Depends(get_current_user)` requiere token válido

📐 Fundamento

Estrategias de versionado:

Recomendado: path versioning.

from fastapi import APIRouter

v1_router = APIRouter(prefix="/v1")
v2_router = APIRouter(prefix="/v2")

@v1_router.get("/pedidos")
def listar_v1(): ...

@v2_router.get("/pedidos")
def listar_v2():  # cambios breaking respecto a v1
    ...

app.include_router(v1_router)
app.include_router(v2_router)

Política de deprecación:

1. Anunciar la nueva versión y la fecha de end-of-life de la vieja (mínimo 6 meses).
2. Header de deprecación en respuestas v1: Deprecation: true + Sunset: Wed, 11 Nov 2026 23:59:59 GMT.
3. Logs de uso de v1: ¿quién la sigue usando?
4. Comunicación directa a usuarios activos de v1.
5. Apagado en la fecha anunciada.

Cambios breaking vs no-breaking:

📐 Fundamento

Paginación:

@app.get("/pedidos")
def listar(page: int = 1, size: int = 20):
    items = db.get_pedidos(offset=(page-1)*size, limit=size)
    total = db.count_pedidos()
    return {
        "items": items,
        "page": page,
        "size": size,
        "total": total,
        "pages": (total + size - 1) // size
    }

Cursor-based pagination (mejor para datasets grandes):

@app.get("/pedidos")
def listar(cursor: Optional[str] = None, size: int = 20):
    items = db.get_pedidos_after(cursor=cursor, limit=size+1)
    next_cursor = items[-1].id if len(items) > size else None
    return {"items": items[:size], "next_cursor": next_cursor}

Rate limiting:

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/pedidos")
@limiter.limit("10/minute")
def crear_pedido(...): ...

Manejo consistente de errores:

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={
            "error": "validation_error",
            "message": "Inputs inválidos",
            "details": exc.errors(),
            "request_id": request.headers.get("X-Request-ID")
        }
    )

CORS (Cross-Origin Resource Sharing) — para que el frontend en otro dominio pueda llamar:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://la-esquina.com"],   # NO usar "*" en producción
    allow_credentials=True,
    allow_methods=["GET", "POST", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"]
)