Examples

Copy-paste ready snippets

Real, complete integrations across every supported framework, storage backend, and SaaS pattern. Every card is self-contained and works as-is.

Framework quick-starts

Drop-in setup for every supported framework. Each snippet is runnable as-is with an in-memory store.

PythonFastAPIMiddlewareDepends()

FastAPI middleware & Depends()

Global middleware on every request PLUS a stricter Depends() limiter on hot routes like login.

main.pypython

from fastapi import FastAPI, Depends, Request
from halt import RateLimiter, InMemoryStore, presets
from halt.adapters.fastapi import (
    HaltMiddleware,
    create_async_limiter_dependency,
)

# ---- A) Global middleware  every route ----
public = RateLimiter(store=InMemoryStore(), policy=presets.PUBLIC_API)

app = FastAPI()
app.add_middleware(HaltMiddleware, limiter=public)


# ---- B) Stricter Depends() on hot routes ----
strict = RateLimiter(store=InMemoryStore(), policy=presets.AUTH_ENDPOINTS)
auth_limit = create_async_limiter_dependency(strict)


@app.post("/login", dependencies=[Depends(auth_limit)])
async def login(request: Request):
    return {"ok": True}


@app.get("/health")
async def health():
    # Health checks are exempt automatically; middleware skips this route.
    return {"status": "ok"}

PythonFlaskExtensionDecorator

Flask extension & per-route decorator

HaltExtension for the whole app, plus a small `@rate_limited` decorator for per-route overrides.

app.pypython

from functools import wraps
from flask import Flask, request, jsonify
from halt import RateLimiter, InMemoryStore, presets
from halt.adapters.flask import HaltExtension

app = Flask(__name__)

# ---- Global limiter via HaltExtension ----
public_limiter = RateLimiter(store=InMemoryStore(), policy=presets.PUBLIC_API)
halt = HaltExtension(public_limiter)
halt.init_app(app)

# ---- Per-route override via a small decorator ----
strict_limiter = RateLimiter(store=InMemoryStore(), policy=presets.AUTH_ENDPOINTS)


def rate_limited(limiter):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            decision = limiter.check(request)
            if not decision.allowed:
                resp = jsonify(
                    error="too_many_requests",
                    retry_after=decision.retry_after,
                )
                resp.status_code = 429
                return resp
            return fn(*args, **kwargs)
        return wrapper
    return decorator


@app.get("/api/data")
def data():
    return {"ok": True}


@app.post("/login")
@rate_limited(strict_limiter)
def login():
    return {"ok": True}

PythonDjangoMiddleware

Django middleware setup

Declarative Halt middleware + HALT settings block. Zero code in your views.

settings.pypython

# Django settings.py
from halt import presets

MIDDLEWARE = [
    # ... your other middleware
    "halt.adapters.django.HaltMiddleware",
]

HALT = {
    "store": "redis",           # or "memory", "postgres", "mongodb", …
    "policy": "public_api",     # named preset OR a Policy instance below
}

# For a custom policy, drop it in as HALT_POLICY:
HALT_POLICY = presets.PUBLIC_API

PythonDjangoDRFThrottle

Django REST Framework throttle class

A small BaseThrottle subclass backed by a halt-rate limiter apply via `throttle_classes` on any ViewSet.

throttles.pypython

# throttles.py  a small DRF throttle wrapping halt-rate
from rest_framework.throttling import BaseThrottle
from halt import RateLimiter, InMemoryStore, presets

# Share one limiter across requests. Use RedisStore in production.
_limiter = RateLimiter(store=InMemoryStore(), policy=presets.PUBLIC_API)


class HaltThrottle(BaseThrottle):
    """DRF-native throttle backed by a halt-rate limiter."""

    def _client_ip(self, request):
        xff = request.META.get("HTTP_X_FORWARDED_FOR")
        if xff:
            return xff.split(",")[0].strip()
        return request.META.get("REMOTE_ADDR", "")

    def allow_request(self, request, view):
        decision = _limiter.check(self._client_ip(request))
        if not decision.allowed:
            self._retry_after = decision.retry_after
            return False
        return True

    def wait(self):
        return getattr(self, "_retry_after", None)


# views.py  attach to any ViewSet
from rest_framework.viewsets import ViewSet
from rest_framework.response import Response
from .throttles import HaltThrottle


class HelloViewSet(ViewSet):
    throttle_classes = [HaltThrottle]

    def list(self, request):
        return Response({"message": "Hello, throttled world!"})

TypeScriptExpressMiddleware

Express global + per-route limits

Apply Halt globally with app.use, then attach a stricter limiter to sensitive routes like /login.

index.tstypescript

import express from 'express';
import { RateLimiter, InMemoryStore, presets } from 'halt-rate';
import { haltMiddleware } from 'halt-rate/express';

const app = express();

// ---- Global limit ----
const publicLimiter = new RateLimiter({
  store: new InMemoryStore(),
  policy: presets.PUBLIC_API,
});
app.use(haltMiddleware(publicLimiter));

// ---- Tighter per-route limit ----
const authLimiter = new RateLimiter({
  store: new InMemoryStore(),
  policy: presets.AUTH_ENDPOINTS,
});

app.post('/login', haltMiddleware(authLimiter), (req, res) => {
  // login logic…
  res.json({ ok: true });
});

app.get('/', (_req, res) => res.json({ hello: 'world' }));
app.listen(3000);

TypeScriptNext.jsEdge

Next.js middleware.ts + route handler

Global middleware for /api/*, or wrap a single route handler with withHalt for a per-route policy.

middleware.tstypescript

// middleware.ts
import { NextResponse } from 'next/server';
import type { NextRequest } from 'next/server';
import { RateLimiter, InMemoryStore, presets } from 'halt-rate';
import { nextMiddleware } from 'halt-rate/next';

const limiter = new RateLimiter({
  store: new InMemoryStore(),
  policy: presets.PUBLIC_API,
});

export async function middleware(request: NextRequest) {
  const result = await nextMiddleware(limiter)(request);
  if (result instanceof NextResponse) return result;
  return NextResponse.next();
}

export const config = { matcher: '/api/:path*' };


// app/api/reports/route.ts  per-route override
import { withHalt } from 'halt-rate/next';
import { presets } from 'halt-rate';

export const GET = withHalt(
  async () => Response.json({ report: 'ok' }),
  { policy: presets.EXPENSIVE_OPS },
);

TypeScriptHonoEdgeCloudflare

Hono edge middleware

Runs on Node, Bun, Deno, Cloudflare Workers, and Vercel Edge. Client IP is auto-detected from edge headers.

src/index.tstypescript

import { Hono } from 'hono';
import { RateLimiter, InMemoryStore, presets } from 'halt-rate';
import { haltMiddleware } from 'halt-rate/hono';

const limiter = new RateLimiter({
  store: new InMemoryStore(),
  policy: presets.PUBLIC_API,
});

const app = new Hono();

// Deploy on Cloudflare Workers  the adapter reads x-forwarded-for,
// cf-connecting-ip, or x-real-ip by default. Override with getClientIp.
app.use('*', haltMiddleware({ limiter }));

app.get('/', (c) => c.json({ message: 'Hello, rate-limited world!' }));

export default app;

Storage backends

Swap InMemoryStore for a production backend. Redis is recommended for anything running more than one process.

TypeScriptRedisExpressProduction

Redis atomic distributed limits

Recommended for production. Every algorithm runs as a single-key Lua script inside Redis, so bursts never over-admit.

index.tstypescript

import express from 'express';
import Redis from 'ioredis';
import { RateLimiter, RedisStore, presets } from 'halt-rate';
import { haltMiddleware } from 'halt-rate/express';

const store = new RedisStore({
  client: new Redis(process.env.REDIS_URL),
  failMode: 'open',           // 'closed' = 429 when Redis is unreachable
  onError: (err) => console.error('halt.redis', err),
});

const limiter = new RateLimiter({ store, policy: presets.PUBLIC_API });

const app = express();
app.use(haltMiddleware(limiter));
app.get('/', (_req, res) => res.json({ ok: true }));
app.listen(3000);

PythonFastAPIRedisAsync

FastAPI + async Redis

Async store paired with acheck() the FastAPI middleware awaits it internally so nothing else changes.

app/main.pypython

import redis.asyncio as aioredis
from fastapi import FastAPI
from halt import RateLimiter, AsyncRedisStore, presets
from halt.adapters.fastapi import HaltMiddleware

store = AsyncRedisStore(
    client=aioredis.Redis.from_url("redis://localhost:6379"),
    fail_mode="open",
)

limiter = RateLimiter(store=store, policy=presets.PUBLIC_API)

app = FastAPI()
app.add_middleware(HaltMiddleware, limiter=limiter)


@app.get("/")
async def root():
    return {"message": "Hello, rate-limited world!"}

PythonPostgreSQLPython

PostgreSQL store

Durable, relational limit storage. Halt creates the table automatically on first use.

main.pypython

from halt import RateLimiter, presets
from halt.stores.postgres import PostgresStore
from halt.adapters.fastapi import HaltMiddleware
from fastapi import FastAPI

store = PostgresStore(
    connection_string="postgresql://user:pass@localhost/db",
    table_name="rate_limits",
    min_size=1,
    max_size=10,
)

limiter = RateLimiter(store=store, policy=presets.PUBLIC_API)

app = FastAPI()
app.add_middleware(HaltMiddleware, limiter=limiter)

TypeScriptMongoDBExpress

MongoDB store

Document storage with TTL indexes. Good option if MongoDB is already in your stack.

index.tstypescript

import express from 'express';
import { RateLimiter, presets } from 'halt-rate';
import { MongoDBStore } from 'halt-rate/stores/mongodb';
import { haltMiddleware } from 'halt-rate/express';

const store = new MongoDBStore({
  connectionString: 'mongodb://localhost:27017',
  database: 'halt',
  collection: 'rate_limits',
});

const limiter = new RateLimiter({ store, policy: presets.PUBLIC_API });

const app = express();
app.use(haltMiddleware(limiter));
app.listen(3000);

TypeScriptCloudflareUpstashEdge

Cloudflare Workers + Upstash Redis

Distributed limits on the edge inject a fetch-based Redis client into RedisStore.

src/worker.tstypescript

import { Redis } from '@upstash/redis';
import { Hono } from 'hono';
import { RateLimiter, RedisStore, presets } from 'halt-rate';
import { haltMiddleware } from 'halt-rate/hono';

const upstash = Redis.fromEnv();

// Thin adapter satisfying RedisClientLike  only the methods your
// algorithms need. All four algorithms use eval + TIME.
const client = {
  eval: (script: string, keys: string[], args: (string | number)[]) =>
    upstash.eval(script, keys, args),
  // …get, set, del, hget, etc. as needed
};

const store = new RedisStore({ client });
const limiter = new RateLimiter({ store, policy: presets.PUBLIC_API });

const app = new Hono();
app.use('*', haltMiddleware({ limiter }));

export default app;

SaaS patterns

Multi-tenant limits, plan-based tiers, quotas, and abuse controls.

TypeScriptSaaSAPI keysPlansDynamic limits

API-key auth → per-tenant plan (live-updatable)

Extract the API key, look up the tenant + plan, apply the matching PLAN_* preset. Uses cachedPolicyResolver so plan changes propagate fleet-wide within the TTL no restart.

src/index.tstypescript

import express from 'express';
import Redis from 'ioredis';
import {
  RateLimiter,
  RedisStore,
  cachedPolicyResolver,
  presets,
  getPlanPolicy,
} from 'halt-rate';
import { haltMiddleware } from 'halt-rate/express';

const redis = new Redis(process.env.REDIS_URL);
const store = new RedisStore({ client: redis, failMode: 'open' });

// Your source of truth  a real app hits its DB + a cache.
async function lookupTenant(apiKey?: string) {
  if (!apiKey) return null;
  const raw = await redis.get(`tenant:by_key:${apiKey}`);
  return raw ? (JSON.parse(raw) as { id: string; plan: string }) : null;
}

// Policy per request, cached per API key for 5s.
// Change a tenant's plan in Redis → every server sees it within 5s.
const policy = cachedPolicyResolver(
  async (req) => {
    const tenant = await lookupTenant(req.header('x-api-key'));
    return getPlanPolicy(tenant?.plan ?? 'free') ?? presets.PLAN_FREE;
  },
  {
    ttlMs: 5_000,
    key: (req) => req.header('x-api-key') ?? 'anonymous',
  },
);

const limiter = new RateLimiter({ store, policy });

const app = express();

app.use((req, res, next) => {
  if (!req.header('x-api-key')) {
    return res.status(401).json({ error: 'missing_api_key' });
  }
  return haltMiddleware(limiter)(req, res, next);
});

app.get('/v1/data', (_req, res) => res.json({ ok: true }));
app.listen(3000);

PythonQuotasTelemetrySaaS

Quotas + telemetry

Monthly usage cap alongside per-second limits. Share a StatsCollector across the limiter and QuotaManager for unified counters.

app.pypython

from halt import (
    RateLimiter,
    QuotaManager,
    StatsCollector,
    Quota,
    QuotaPeriod,
    presets,
)
from halt.stores.postgres import PostgresStore

store = PostgresStore(connection_string="postgresql://…")

# One collector for both  unified counters in one snapshot.
stats = StatsCollector(top_n=20, max_tracked_keys=10_000)

limiter = RateLimiter(store=store, policy=presets.PLAN_PRO, telemetry=stats)
quotas = QuotaManager(store=store, telemetry=stats)

MONTHLY = Quota(name="pro_monthly", limit=1_000_000, period=QuotaPeriod.MONTH)


def handle(user_id: str):
    # 1) Per-second/minute rate limit
    if not limiter.check(user_id).allowed:
        return {"error": "rate_limited"}, 429

    # 2) Monthly quota
    allowed, usage = quotas.check_quota(user_id, MONTHLY)
    if not allowed:
        return {"error": "quota_exceeded"}, 402

    quotas.consume(user_id, MONTHLY, amount=1)
    return {"ok": True, "usage": usage}, 200


def get_stats():
    return stats.snapshot()

TypeScriptPenaltyAbuseSaaS

Abuse detection with penalties

Track rate-limit violations, temporarily ban abusive keys, and share telemetry with the rest of the stack.

abuse.tstypescript

import {
  RateLimiter,
  RedisStore,
  PenaltyManager,
  PENALTY_MODERATE,
  StatsCollector,
  presets,
} from 'halt-rate';

const stats = new StatsCollector();

const limiter = new RateLimiter({
  store,
  policy: presets.AUTH_ENDPOINTS,
  telemetry: stats,
});

const penalties = new PenaltyManager(store, PENALTY_MODERATE, {
  telemetry: stats,
});

export async function checkAuth(userId: string) {
  // 1) Already banned?
  const active = await penalties.getPenalty(userId);
  if (penalties.isActive(active)) {
    return { allowed: false, reason: 'suspended' as const };
  }

  // 2) Try the limit.
  const decision = await limiter.check(userId);
  if (!decision.allowed) {
    await penalties.recordViolation(userId);
    return { allowed: false, reason: 'rate_limited' as const };
  }

  return { allowed: true };
}

Advanced patterns

Change limits at runtime, weighted endpoints, and observability.

PythonDynamic limitsPolicyRegistryRuntime

Change limits at runtime PolicyRegistry

Register a named policy, hand its resolver to the limiter, mutate it later. The next check sees the new limit no restart.

runtime.pypython

from halt import RateLimiter, PolicyRegistry, presets

registry = PolicyRegistry([presets.PUBLIC_API])

limiter = RateLimiter(
    store=store,
    policy=registry.resolver(lambda req: "public_api"),
)

# Later  flip the limit live from an admin endpoint.
def bump_limit(new_limit: int):
    registry.update("public_api", limit=new_limit)
    # Raising limit past the old burst? update() recomputes burst automatically.

TypeScriptCostWeightedPolicies

Weighted endpoints cost per operation

Charge expensive routes multiple tokens. Reports show up in StatsCollector.snapshot() under byEndpoint.cost.

index.tstypescript

import express from 'express';
import { RateLimiter, InMemoryStore, presets } from 'halt-rate';
import { haltMiddleware } from 'halt-rate/express';

const limiter = new RateLimiter({
  store: new InMemoryStore(),
  policy: presets.PUBLIC_API,
});

const app = express();

// Cheap route  cost defaults to 1.
app.get(
  '/v1/data',
  haltMiddleware(limiter, { cost: 1 }),
  (_req, res) => res.json({ ok: true }),
);

// Expensive report  charge 10 tokens per call.
app.get(
  '/v1/reports/summary',
  haltMiddleware(limiter, { cost: 10 }),
  (_req, res) => res.json({ report: '…' }),
);

app.listen(3000);

PythonObservabilityStatsCollectorFastAPI

Live /halt/stats endpoint

Serve StatsCollector.snapshot() on a small admin endpoint instant dashboard-ready JSON.

app.pypython

from fastapi import FastAPI, Depends
from halt import RateLimiter, StatsCollector, presets
from halt.adapters.fastapi import HaltMiddleware

stats = StatsCollector(top_n=20, max_tracked_keys=10_000)

limiter = RateLimiter(
    store=store,
    policy=presets.PUBLIC_API,
    telemetry=stats,
)

app = FastAPI()
app.add_middleware(HaltMiddleware, limiter=limiter)


def require_admin():
    # replace with your real auth
    ...


@app.get("/halt/stats", dependencies=[Depends(require_admin)])
def halt_stats():
    return stats.snapshot()   # allowedTotal, blockedTotal, byPolicy, …

TypeScriptOpenTelemetryCompositeObservability

CompositeTelemetry Stats + OpenTelemetry

Local StatsCollector for the admin endpoint AND OTel counters for fleet-wide dashboards. One hook, both destinations.

telemetry.tstypescript

import { metrics } from '@opentelemetry/api';
import {
  RateLimiter,
  CompositeTelemetry,
  StatsCollector,
  OpenTelemetryMetrics,
  presets,
} from 'halt-rate';

export const stats = new StatsCollector({ topN: 20 });

const telemetry = new CompositeTelemetry([
  stats,
  new OpenTelemetryMetrics(metrics.getMeter('halt')),
]);

export const limiter = new RateLimiter({
  store,
  policy: presets.PUBLIC_API,
  telemetry,
});

// stats.snapshot() → /halt/stats endpoint
// halt.requests / halt.blocked / halt.cost → your OTel pipeline

Building something not covered here?

The interactive builder generates a runnable project for any language + framework + storage + policy combination try it, then customise.

Open the interactive builder