Metered ⏱️

Metered is an enterprise-grade, asynchronous rate limiting and quota management library for Python. Designed for modern SaaS architectures, it supports FastAPI and Flask natively, offering highly precise algorithms, dynamic LLM token costing, and robust distributed state management via Redis.

---

🚀 Features

• 5 Core Rate Limiting Algorithms:
  • Token Bucket (Smooth bursting)
  • Sliding Window Log (High precision, dynamic costs)
  • Sliding Window Counter (Memory-efficient approximation)
  • Fixed Window (Standard quota tracking)
  • Leaky Bucket (Strict egress shaping)

📊 Detailed Comparison of the Algorithms

Algorithm	Traffic Handling	Memory Usage	Implementation Complexity	Best Used For
Token Bucket	Allows controlled bursts	Low (stores 2 numbers per client)	Moderate	General-purpose APIs
Leaky Bucket	Enforces a smooth, fixed output rate	Low (bounded queue size)	Moderate	Data egress or traffic shaping
Fixed Window Counter	Prone to boundary bursts	Minimal (1 integer counter)	Very Low	Simple internal or low-scale apps
Sliding Window Log	Perfectly accurate	Very High (stores every timestamp)	High	High-security endpoints (e.g., login, MFA)
Sliding Window Counter	Approximated accuracy, smooths boundaries	Low (stores 2 adjacent window counts)	Moderate	Large distributed systems

• Dynamic Costing: Perfect for GenAI/LLM wrappers—calculate the cost (tokens) of a request dynamically at runtime.
• Quota Persistence: Define limits like "10,000 tokens per month" that survive app restarts and synchronize globally via Redis.
• Event-Driven Architecture: Native webhook/event dispatcher with DLQ (Dead Letter Queue), exponential backoff, and strict event throttling to prevent spam when quotas are breached.
• Backend Agnostic: Ships with a thread-safe InMemoryBackend for development and an atomic, Lua-powered RedisBackend for high-throughput production.
• Async Native: Built with asyncio from the ground up, guaranteeing non-blocking behavior.

---

📦 Installation

pip install metered

(Coming soon to PyPI)

---

⚡ Quick Start

FastAPI Example

from fastapi import FastAPI, Request
from metered import Metered, Strategy, IdentifierType

app = FastAPI()
# By default, uses the lightweight InMemoryBackend
meter = Metered()

# Helper to identify users
def get_client_ip(req: Request) -> str:
    return req.client.host if req.client else "127.0.0.1"

# Restrict to 5 requests per 10 seconds per IP
@app.get("/api/data")
@meter.limit(
    max_tokens=5, 
    period=10, 
    strategy=Strategy.FIXED_WINDOW, 
    identifier=get_client_ip
)
async def get_data(request: Request):
    return {"data": "Success!"}

Flask Example

from flask import Flask, request, jsonify
from metered import Metered, Strategy

app = Flask(__name__)
meter = Metered()

def get_client_ip(req):
    return req.remote_addr or "127.0.0.1"

@app.route("/api/data", methods=["GET"])
@meter.limit(max_tokens=5, period=10, strategy=Strategy.FIXED_WINDOW, identifier=get_client_ip)
def get_data():
    return jsonify({"data": "Here is your data!"})

---

🧠 Core Concepts

Stacking Decorators

You can combine multiple constraints on a single endpoint. Metered evaluates them top-down (or as passed) and enforces the strictest rule.

@app.get("/search")
@meter.quota(plan_name="pro_plan", identifier=get_user_id)  # Evaluated first
@meter.limit(max_tokens=100, period=60, strategy=Strategy.SLIDING_WINDOW, identifier=get_user_id) # Sustained limit
@meter.limit(max_tokens=10, period=1, strategy=Strategy.SLIDING_WINDOW, identifier=get_user_id)   # Burst limit
async def search(request: Request):
    return {"results": []}

Dynamic Costs (LLMs / AI Apps)

Unlike standard rate limiters where 1 Request = 1 Token, metered allows you to define a cost function.

async def calculate_llm_cost(req: Request) -> int:
    body = await req.json()
    prompt = body.get("prompt", "")
    # e.g., 1 word = ~1.3 tokens
    return max(1, int(len(prompt.split()) * 1.3))

@app.post("/v1/completions")
@meter.limit(
    max_tokens=5000, 
    period=60, 
    strategy=Strategy.TOKEN_BUCKET, 
    identifier=get_api_key, 
    cost=calculate_llm_cost
)
async def generate_text(request: Request):
    return {"text": "AI response..."}

---

🏭 Production Guide

Using Redis (Recommended for Production)

For distributed systems and multi-worker deployments (e.g., Uvicorn/Gunicorn), use the RedisBackend. It relies entirely on atomic Lua scripts to prevent race conditions.

from redis.asyncio import Redis
from metered import Metered, RedisBackend

redis_client = Redis.from_url("redis://localhost:6379", decode_responses=True)
redis_backend = RedisBackend(redis_client)

meter = Metered(backend=redis_backend, quota_backend=redis_backend)

Quota Engine & Plans

Unlike short-lived rate limits, Quotas are persistent billing boundaries (e.g., Monthly API usage).

# During your app startup phase
@app.on_event("startup")
async def startup():
    # Store a persistent quota plan into the backend
    await meter.quotas.set_plan(
        target="user_123",
        plan_name="pro_plan",
        limit=100000,               # 100k tokens
        reset_period="monthly",     # resets on the 1st of every month
        identifier_type=IdentifierType.USER_ID
    )

Event Dispatcher (Alerts & Webhooks)

Metered includes an advanced event dispatcher with two-phase commits, throttling, and a DLQ (Dead Letter Queue) to reliably notify you (e.g. via Slack) when users approach their limits without spamming your network.

# 1. Configure Persistent Outbox
meter.events.configure(
    backend=redis_backend, 
    persist=True, 
    cooldown_seconds=3600 # Only alert once per hour per user
)

# 2. Start the Background Worker
asyncio.create_task(meter.events.start_worker())

# 3. Listen to Quota Warnings
@meter.events.on_quota_warning(threshold=0.8)
async def handle_quota_warning(target: str, plan_name: str, usage_ratio: float, remaining: int):
    # This handler will be retried with exponential backoff if it fails
    await send_slack_alert(f"User {target} is at {usage_ratio*100}% of their {plan_name} plan.")

---

📊 Performance & Benchmarks

Metered is built to sustain massive concurrency. Below is a benchmark result using locust against the fastapi_saas.py example (which includes dynamic Redis Quotas, Rate Limits, and Event Dispatching):

Type     Name                                           # reqs      # fails |    Avg     Min     Max    Med |   req/s  failures/s
--------|---------------------------------------------------------------------------------------------------|--------|-----------
GET      /premium-data                                    7865     0(0.00%) |    104       2     349    100 |  530.32        0.00

(Tested with 100 concurrent headless users. Zero failures or unhandled exceptions).

---

🌍 Global Middleware

If you want to apply limits globally rather than per-route, metered exports standard middlewares.

FastAPI:

from metered.integrations.fastapi import MeteredMiddleware
from metered import Limit, Strategy

# Limits all traffic globally to 100 req/sec
app.add_middleware(
    MeteredMiddleware, 
    meter=meter, 
    limits=[Limit(max_tokens=100, period=1, strategy=Strategy.FIXED_WINDOW)]
)

---

🤝 Contributing

Contributions are highly welcomed! Please check our open issues.

1. Fork the Project
2. Create your Feature Branch (git checkout -b feature/AmazingFeature)
3. Commit your Changes (git commit -m 'Add some AmazingFeature')
4. Run static checks (mypy metered/)
5. Push to the Branch (git push origin feature/AmazingFeature)
6. Open a Pull Request

📄 License

Distributed under the MIT License. See LICENSE for more information.