Middleware

Rate limiting, logging, timing — cross-cutting concerns as pipeline processors.

The Problem

Every endpoint needs logging, rate limiting, timing. Copy-pasting into each handler violates DRY.

The Solution: Pipeline Processors

from evoid import register_processor
from evoid.core import Context
from evoid.core.extend import before, after

# Rate limiter
_calls: dict[str, list] = {}

async def rate_limit(ctx: Context) -> dict:
    name = ctx.intent.name
    import time
    now = time.time()
    window = 60
    max_calls = 100

    _calls.setdefault(name, [])
    _calls[name] = [t for t in _calls[name] if now - t < window]

    if len(_calls[name]) >= max_calls:
        raise Exception(f"Rate limit exceeded for {name}")

    _calls[name].append(now)
    return {"rate_limited": False}

# Logger
async def log_request(ctx: Context) -> dict:
    print(f"[{ctx.intent.name}] started")
    ctx.state["log_start"] = __import__("time").monotonic()
    return {"logged": True}

async def log_response(ctx: Context) -> dict:
    start = ctx.state.get("log_start", 0)
    elapsed = __import__("time").monotonic() - start
    print(f"[{ctx.intent.name}] completed in {elapsed:.3f}s")
    return {"elapsed": elapsed}

register_processor("rate_limit", rate_limit)
register_processor("log_request", log_request)
register_processor("log_response", log_response)

Attach to Routes

# Apply to specific routes
before("GET:/menu", "rate_limit")
before("POST:/orders", "rate_limit")
after("GET:/menu", "log_response")
after("POST:/orders", "log_response")

# Apply to all routes with a pattern
for route in ["GET:/menu", "POST:/orders", "GET:/orders"]:
    before(route, "log_request")

Apply Globally

from evoid.core.extend import before

# Apply to specific intents
before("GET:/menu", "rate_limit")
before("POST:/orders", "rate_limit")
before("GET:/menu", "log_request")
before("POST:/orders", "log_request")

Insert Relative to Processors

from evoid.core.extend import before_processor, after_processor

# Add auth check before validation
before_processor("POST:/orders", "validate", "check_auth")

# Add audit after authorization
after_processor("POST:/orders", "authorize", "audit_log")

Replace Entire Pipeline

from evoid.core.extend import replace_pipeline

# For health checks — skip everything
replace_pipeline("GET:/health", ["handle_health"])

What You Learned

Next: Status Codes

Let’s handle HTTP status codes — Status Codes.