traffik 1.1.0

Asynchronous distributed rate limiting for FastAPI/Starlette applications.

Traffik

Asynchronous distributed rate limiting for FastAPI/Starlette applications.

Features

• Fully Asynchronous: Built for async/await with non-blocking operations and minimal overhead
• Distributed-First: Atomic operations with distributed locks (Redis, Memcached) achieving high accuracy under concurrency
• 10+ Strategies: Fixed Window, Sliding Window (Log & Counter), Token Bucket, Leaky Bucket, GCRA, and more
• HTTP & WebSocket: Full-featured rate limiting for both protocols with per-message throttling support
• Production-Ready: Circuit breakers, automatic retries, backend failover, and custom error handling
• Flexible Integration: Dependencies, decorators, middleware, or direct calls — use what fits your architecture
• Highly Extensible: Simple, well-documented APIs for custom backends, strategies, error handlers, and identifiers
• Observable: Rich error context and strategy statistics for monitoring
• Performance-Optimized: Lock striping, connection pooling, script caching, and minimal memory footprint

Built for production workloads.

Documentation

Full documentation is available at https://ti-oluwa.github.io/traffik/, covering:

• Getting Started - up and running in minutes
• Core Concepts - backends, strategies, identifiers, rate formats
• Integration - dependencies, decorators, middleware, direct usage
• Advanced Features - response headers, throttle rules, quota contexts, and more
• Configuration - lock settings, environment variables
• Error Handling - retries, circuit breakers, fallbacks
• Extending Traffik - custom backends and strategies
• Testing - patterns for testing throttled endpoints
• Benchmarks - performance comparisons

Installation

# Basic (InMemory backend only)
pip install traffik

# With Redis support
pip install "traffik[redis]"

# With Memcached support
pip install "traffik[memcached]"

# All backends
pip install "traffik[all]"

Quick Start

from fastapi import FastAPI, Depends
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend

backend = InMemoryBackend(namespace="myapp")
app = FastAPI(lifespan=backend.lifespan)

throttle = HTTPThrottle("api:v1", rate="100/min", backend=backend)

@app.get("/items", dependencies=[Depends(throttle)])
async def list_items():
    return {"items": []}

See the Quick Start guide for more examples including Redis, WebSocket throttling, middleware setup, and advanced patterns.

License

MIT — see LICENSE for details.