← Volver al blog
· 3 min de lectura · 446 palabras

Starlette y middlewares a fondo: logging, rate limiting, CORS

Introducción

Starlette es uno de los frameworks ASGI más infravalorados. FastAPI está construido sobre él, pero Starlette por sí solo es potente, rápido y minimalista. Entenderlo te da control total sobre tu aplicación web.

¿Qué es Starlette?

Starlette es un framework ASGI ligero para Python. Pip install starlette y tienes routing, middleware, WebSocket, GraphQL, testing, y más.

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def homepage(request):
    return JSONResponse({"message": "Hola, mundo"})

app = Starlette(routes=[
    Route("/", endpoint=homepage)
])

Middleware personalizado

Logging de peticiones

import time
from starlette.middleware.base import BaseHTTPMiddleware

class RequestLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.time()
        response = await call_next(request)
        elapsed = time.time() - start
        print(f"{request.method} {request.url.path} - {elapsed:.4f}s")
        return response

app.add_middleware(RequestLogMiddleware)

Limitación de tasa (rate limiting)

from collections import defaultdict
import time

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, max_requests=100, window=60):
        super().__init__(app)
        self.max_requests = max_requests
        self.window = window
        self.requests = defaultdict(list)

    async def dispatch(self, request, call_next):
        client_ip = request.client.host
        now = time.time()
        
        # Limpiar entradas viejas
        self.requests[client_ip] = [
            t for t in self.requests[client_ip]
            if now - t < self.window
        ]
        
        if len(self.requests[client_ip]) >= self.max_requests:
            return JSONResponse(
                {"error": "Rate limit exceeded"},
                status_code=429
            )
        
        self.requests[client_ip].append(now)
        return await call_next(request)

CORS personalizado

from starlette.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://ejemplo.com"],
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization"],
    max_age=3600  # Cache preflight 1 hora
)

Manejo de errores global

from starlette.exceptions import HTTPException
from starlette.requests import Request

async def error_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        {"error": exc.detail, "path": request.url.path},
        status_code=exc.status_code
    )

app.add_exception_handler(HTTPException, error_handler)

Subidas de archivos con streaming

from starlette.datastructures import UploadFile

async def upload_file(request):
    form = await request.form()
    file: UploadFile = form["file"]
    
    # Streaming directo a disco sin cargar en memoria
    with open(f"uploads/{file.filename}", "wb") as f:
        while chunk := await file.read(1024 * 1024):  # 1MB chunks
            f.write(chunk)
    
    return JSONResponse({"file": file.filename})

WebSockets

from starlette.endpoints import WebSocketEndpoint
from starlette.routing import WebSocketRoute

class EchoEndpoint(WebSocketEndpoint):
    async def on_receive(self, websocket, data):
        await websocket.send_text(f"Echo: {data}")

routes = [
    WebSocketRoute("/ws", endpoint=EchoEndpoint)
]

Background tasks

from starlette.background import BackgroundTask

async def send_email_task(user_id):
    # Lógica de envío de email
    pass

async def create_user(request):
    data = await request.json()
    user = await save_user(data)
    task = BackgroundTask(send_email_task, user.id)
    return JSONResponse({"id": user.id}, background=task)

Testing integrado

from starlette.testclient import TestClient

def test_homepage():
    client = TestClient(app)
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hola, mundo"}

Conclusión

Starlette es la base perfecta para construir APIs. Es minimalista, rápido y bien diseñado. Antes de añadir FastAPI a tu proyecto, considera si realmente necesitas todo lo que FastAPI añade (validación automática, OpenAPI, etc.) o si Starlette es suficiente. Muchas veces lo es.