1. Token Bucket

Best for: APIs with burst allowances

rule = RateThrottleRule(
    name='api_burst',
    limit=100,
    window=60,
    strategy='token_bucket',
    burst=150  # Allow up to 150 tokens in bucket
)

2. Leaky Bucket

Best for: Smooth, constant rate processing

rule = RateThrottleRule(
    name='api_steady',
    limit=100,
    window=60,
    strategy='leaky_bucket'
)

3. Fixed Window

Best for: Simple, efficient rate limiting

rule = RateThrottleRule(
    name='api_window',
    limit=100,
    window=60,
    strategy='fixed_window'
)

4. Sliding Window Log

Best for: Precise rate limiting without edge cases

rule = RateThrottleRule(
    name='api_precise',
    limit=100,
    window=60,
    strategy='sliding_window'
)

4. Sliding Window Counter

Best for: Accurate limiting with efficient memory usage

rule = RateThrottleRule(
    name='api_default',
    limit=100,
    window=60,
    strategy='sliding_counter'
)

from ratethrottle import create_limiter

# Using Redis for distributed rate limiting
limiter = create_limiter('redis', 'redis://localhost:6379/0')

# Now works across multiple servers!

Create ratethrottle.yaml:

# Storage backend
storage:
  type: redis
  redis:
    host: localhost
    port: 6379
    db: 0

# Global settings
global:
  enabled: true
  default_strategy: sliding_counter
  headers_enabled: true

# Rate limiting rules
rules:
  - name: api_default
    limit: 1000
    window: 3600
    strategy: token_bucket
    
  - name: auth_strict
    limit: 5
    window: 60
    strategy: sliding_window
    block_duration: 900

# DDoS Protection
ddos_protection:
  enabled: true
  threshold: 10000
  auto_block: true
  block_duration: 3600

Load configuration:

from ratethrottle import ConfigManager, RateThrottleCore

config = ConfigManager('ratethrottle.yaml')
limiter = RateThrottleCore()

for rule in config.get_rules():
    limiter.add_rule(rule)

from concurrent import futures
import grpc
from ratethrottle import GRPCRateLimitInterceptor, GRPCLimits
        
 # Create interceptor
interceptor = GRPCRateLimitInterceptor(
     GRPCLimits(
        requests_per_minute=100,
        concurrent_requests=10
            )
)

# Create server with rate limiting
server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[interceptor]
)

from ratethrottle import GraphQLRateLimiter, GraphQLLimits

limiter = GraphQLRateLimiter(
    GraphQLLimits(
        queries_per_minute=100,
        max_complexity=1000
    )
)

# Check if query is allowed
error = limiter.check_rate_limit(
    document_ast=parsed_query,
    context=request_context
)
    
if error:
    raise error # GraphQLError

from ratethrottle import WebSocketRateLimiter, WebSocketLimits

limiter = WebSocketRateLimiter(
    WebSocketLimits(
        connections_per_minute=10,
        messages_per_minute=100
    )
)

# Check if connection allowed
if await limiter.check_connection("client_id"):
    # Accept connection
    await limiter.register_connection("client_id", websocket)

from ratethrottle.monitoring import RateThrottleMonitor
from ratethrottle.alerting import AlertDispatcher

monitor = RateThrottleMonitor(
    {
        'enabled': True,
        'interval': 60,
        'log_metrics': True,
        'export_json': True,
        'export_path': 'metrics/metrics.json',
    },
    limiter=limiter,
    ddos=ddos,
    analytics=analytics,
)

dispatcher = AlertDispatcher(
    {
        'enabled': True,
        'cooldown_seconds': 300,
        'thresholds': {
            'block_rate_warning': 5.0,
            'block_rate_critical': 20.0,
            'violations_per_minute_warning': 50.0,
            'violations_per_minute_critical': 200.0,
            'ddos_score_warning': 0.5,
            'ddos_score_critical': 0.8,
        },
        'slack': {
            'enabled': True,
            'channel': '#alerts',
            'username': 'RateThrottle',
        },
        'webhook': {
            'enabled': False,
            'url': '',
            'timeout': 10,
        },
    }
)

monitor.start()

snapshot = monitor.snapshot_now()
dispatcher.check_and_alert(snapshot)

from ratethrottle import DDoSProtection

ddos = DDoSProtection({
    'enabled': True,
    'threshold': 10000,
    'window': 60,
    'auto_block': True,
    'block_duration': 3600
})

# Analyze traffic pattern
pattern = ddos.analyze_traffic(
    identifier='192.168.1.100',
    endpoint='/api/data'
)

if pattern.is_suspicious:
    print(f"⚠️ Suspicious activity detected!")
    print(f"Request rate: {pattern.request_rate:.2f} req/s")
    print(f"Suspicion score: {pattern.suspicious_score:.2f}")

# Add to whitelist (bypass all limits)
limiter.add_to_whitelist('192.168.1.100')

# Add to blacklist (block all requests)
limiter.add_to_blacklist('192.168.1.200', duration=3600)  # Block for 1 hour

# Remove from blacklist
limiter.remove_from_blacklist('192.168.1.200')

def handle_violation(violation):
    """Custom violation handler"""
    print(f"⚠️ Violation: {violation.identifier}")
    print(f"Rule: {violation.rule_name}")
    print(f"Requests: {violation.requests_made}/{violation.limit}")
    
    # Send alert, log to database, etc.

limiter.register_violation_callback(handle_violation)

# Get metrics
metrics = limiter.get_metrics()

print(f"Total requests: {metrics['total_requests']}")
print(f"Blocked requests: {metrics['blocked_requests']}")
print(f"Block rate: {metrics['block_rate']:.2f}%")
print(f"Recent violations: {len(metrics['recent_violations'])}")

# Reset metrics
limiter.reset_metrics()

RateThrottle includes a powerful CLI for monitoring and management:

# Start interactive monitoring dashboard
ratethrottle monitor --config ratethrottle.yaml

# Test rate limiting configuration
ratethrottle test --rule api_default --identifier 192.168.1.100 --requests 150

# Manage whitelist/blacklist
ratethrottle manage --blacklist-add 192.168.1.50 --duration 3600
ratethrottle manage --whitelist-add 10.0.0.5

# View configuration
ratethrottle config --show

# Validate configuration
ratethrottle config --validate

# Export statistics
ratethrottle stats --export report.json