import psutil
import time
 
 
class AdaptiveRateLimiter:
    """
    Adjusts effective limit based on system CPU and memory utilization.
    Reduces limits when system is under stress. Restores when healthy.
    """
 
    def __init__(self, base_limit: int, redis_limiter):
        self.base_limit = base_limit
        self.redis_limiter = redis_limiter
 
    def get_effective_limit(self) -> int:
        cpu_percent = psutil.cpu_percent(interval=0.1)
        memory_percent = psutil.virtual_memory().percent
 
        if cpu_percent > 90 or memory_percent > 90:
            # Critical: reduce to 20% of normal
            return max(1, int(self.base_limit * 0.2))
        elif cpu_percent > 75 or memory_percent > 75:
            # High load: reduce to 50%
            return max(1, int(self.base_limit * 0.5))
        elif cpu_percent > 60 or memory_percent > 60:
            # Moderate: reduce to 75%
            return max(1, int(self.base_limit * 0.75))
        else:
            # Normal: full limit
            return self.base_limit
 
    def is_allowed(self, identifier: str) -> bool:
        effective_limit = self.get_effective_limit()
        return self.redis_limiter.is_allowed(identifier, limit_override=effective_limit)

import collections
import threading
 
 
class ErrorRateAdaptiveLimiter:
    """
    Reduces rate limits when downstream error rates are high.
    Implements a sliding window error tracker.
    """
 
    def __init__(self, base_limit: int):
        self.base_limit = base_limit
        self.error_window = collections.deque()
        self.success_window = collections.deque()
        self.lock = threading.Lock()
        self.window_size = 60  # 1 minute
 
    def record_response(self, is_error: bool) -> None:
        now = time.time()
        with self.lock:
            window_start = now - self.window_size
            # Evict old records
            while self.error_window and self.error_window[0] < window_start:
                self.error_window.popleft()
            while self.success_window and self.success_window[0] < window_start:
                self.success_window.popleft()
 
            if is_error:
                self.error_window.append(now)
            else:
                self.success_window.append(now)
 
    def get_effective_limit(self) -> int:
        with self.lock:
            total = len(self.error_window) + len(self.success_window)
            if total == 0:
                return self.base_limit
 
            error_rate = len(self.error_window) / total
 
            if error_rate > 0.5:    # >50% errors
                return max(1, int(self.base_limit * 0.1))
            elif error_rate > 0.25: # >25% errors
                return max(1, int(self.base_limit * 0.5))
            elif error_rate > 0.10: # >10% errors
                return max(1, int(self.base_limit * 0.75))
            else:
                return self.base_limit

class BehavioralRateLimiter:
    """
    Reduces limits for users showing suspicious patterns.
    Increases limits for trusted, verified users.
    """
 
    TRUST_LEVELS = {
        "anonymous":  0.2,   # 20% of base limit
        "new_user":   0.5,   # 50% of base limit
        "verified":   1.0,   # 100% of base limit
        "premium":    2.0,   # 200% of base limit
        "enterprise": 5.0,   # 500% of base limit
    }
 
    def __init__(self, base_limit: int, redis_limiter):
        self.base_limit = base_limit
        self.redis_limiter = redis_limiter
 
    def get_limit_for_user(self, user_id: str, trust_level: str) -> int:
        multiplier = self.TRUST_LEVELS.get(trust_level, 0.2)
        return max(1, int(self.base_limit * multiplier))
 
    def is_allowed(self, user_id: str, trust_level: str) -> bool:
        limit = self.get_limit_for_user(user_id, trust_level)
        return self.redis_limiter.is_allowed(user_id, limit_override=limit)

from graphql import parse, DocumentNode
from graphql.language.ast import FieldNode, SelectionSetNode
 
 
class GraphQLCostAnalyzer:
    """
    Calculates the cost of a GraphQL query for rate limiting purposes.
    """
 
    # Cost per field type
    FIELD_COSTS = {
        "users":        10,  # returns list
        "repositories": 5,
        "commits":      3,
        "viewer":       0,   # cheap, returns current user
        "_default":     1
    }
 
    # Multiplier for list fields (per item)
    LIST_MULTIPLIERS = {
        "users":        100,  # could return up to 100 users
        "repositories": 50,
        "commits":      30,
    }
 
    def calculate_cost(self, query: str) -> int:
        try:
            document = parse(query)
            return self._calculate_document_cost(document)
        except Exception:
            return 1000  # Unparseable query gets high cost
 
    def _calculate_document_cost(self, document: DocumentNode) -> int:
        total_cost = 0
        for definition in document.definitions:
            if hasattr(definition, "selection_set"):
                total_cost += self._calculate_selection_cost(
                    definition.selection_set, depth=0
                )
        return max(1, total_cost)
 
    def _calculate_selection_cost(
        self, selection_set: SelectionSetNode, depth: int
    ) -> int:
        if not selection_set or depth > 10:
            return 0
 
        cost = 0
        for selection in selection_set.selections:
            if isinstance(selection, FieldNode):
                field_name = selection.name.value
                field_cost = self.FIELD_COSTS.get(
                    field_name, self.FIELD_COSTS["_default"]
                )
                multiplier = self.LIST_MULTIPLIERS.get(field_name, 1)
                cost += field_cost * multiplier
 
                if selection.selection_set:
                    cost += self._calculate_selection_cost(
                        selection.selection_set, depth + 1
                    )
        return cost
 
 
# Usage in FastAPI/Flask middleware
analyzer = GraphQLCostAnalyzer()
 
def check_graphql_rate_limit(user_id: str, query: str, token_bucket) -> dict:
    cost = analyzer.calculate_cost(query)
    result = token_bucket.is_allowed(user_id, cost=cost)
    result["query_cost"] = cost
    return result

# Define weights for different endpoints and methods
ENDPOINT_WEIGHTS = {
    ("GET",  "/api/users"):              1,
    ("GET",  "/api/users/{id}"):         1,
    ("POST", "/api/users"):              2,
    ("GET",  "/api/reports"):            5,
    ("POST", "/api/reports/generate"):   50,
    ("GET",  "/api/export"):             25,
    ("POST", "/api/bulk"):               10,   # per item in bulk
    "_default":                          1
}
 
def get_request_cost(method: str, path: str, body: dict = None) -> int:
    key = (method, path)
    cost = ENDPOINT_WEIGHTS.get(key, ENDPOINT_WEIGHTS["_default"])
 
    # Bulk operations: cost multiplied by item count
    if path == "/api/bulk" and body and "items" in body:
        cost = cost * len(body["items"])
 
    return cost

from dataclasses import dataclass
from typing import Optional
 
 
@dataclass
class RateLimitTier:
    name: str
    requests_per_second: int
    requests_per_minute: int
    requests_per_hour: int
    requests_per_day: int
    requests_per_month: Optional[int]
    max_burst: int
    concurrent_connections: int
    max_response_size_mb: int
 
 
TIERS = {
    "free": RateLimitTier(
        name="free",
        requests_per_second=1,
        requests_per_minute=30,
        requests_per_hour=500,
        requests_per_day=2_000,
        requests_per_month=50_000,
        max_burst=5,
        concurrent_connections=2,
        max_response_size_mb=1
    ),
    "starter": RateLimitTier(
        name="starter",
        requests_per_second=10,
        requests_per_minute=300,
        requests_per_hour=5_000,
        requests_per_day=50_000,
        requests_per_month=500_000,
        max_burst=50,
        concurrent_connections=10,
        max_response_size_mb=10
    ),
    "pro": RateLimitTier(
        name="pro",
        requests_per_second=100,
        requests_per_minute=3_000,
        requests_per_hour=50_000,
        requests_per_day=500_000,
        requests_per_month=5_000_000,
        max_burst=500,
        concurrent_connections=50,
        max_response_size_mb=50
    ),
    "enterprise": RateLimitTier(
        name="enterprise",
        requests_per_second=1_000,
        requests_per_minute=30_000,
        requests_per_hour=500_000,
        requests_per_day=None,   # Unlimited daily
        requests_per_month=None,  # Unlimited monthly
        max_burst=5_000,
        concurrent_connections=500,
        max_response_size_mb=100
    )
}

import redis
import time
 
 
class TieredRateLimiter:
 
    SCRIPT = """
local results = {}
for i, key in ipairs(KEYS) do
    local limit = tonumber(ARGV[i * 2 - 1])
    local window = tonumber(ARGV[i * 2])
    local now = tonumber(ARGV[#ARGV])
    local window_id = math.floor(now / window)
    local full_key = key .. ':' .. window_id
    local count = redis.call('INCR', full_key)
    if count == 1 then
        redis.call('EXPIRE', full_key, window * 2)
    end
    if count > limit then
        return {0, i, limit, count}  -- denied at tier i
    end
    table.insert(results, limit - count)
end
return {1, 0, 0, 0}  -- all limits passed
"""
 
    def __init__(self, r: redis.Redis):
        self.r = r
        self._script = r.register_script(self.SCRIPT)
 
    def is_allowed(self, user_id: str, tier: str) -> dict:
        tier_config = TIERS.get(tier, TIERS["free"])
        now = int(time.time())
 
        # Check all applicable time-window limits
        limits_to_check = [
            (tier_config.requests_per_second, 1),
            (tier_config.requests_per_minute, 60),
            (tier_config.requests_per_hour, 3600),
        ]
        if tier_config.requests_per_day:
            limits_to_check.append((tier_config.requests_per_day, 86400))
 
        keys = [f"rl:{tier}:{user_id}:{window}" for _, window in limits_to_check]
        args = []
        for limit, window in limits_to_check:
            args.extend([limit, window])
        args.append(now)
 
        result = self._script(keys=keys, args=args)
        allowed = bool(int(result[0]))
 
        return {
            "allowed": allowed,
            "tier": tier,
            "denied_at_window": int(result[1]) if not allowed else None
        }

{
  viewer {
    # 1 point
    repositories(first: 100) {
      # 100/100 = 1 point... actually more complex
      nodes {
        name
        issues(first: 50) {
          # adds to cost
          nodes {
            title
          }
        }
      }
    }
  }
}
# Total: ~51 points

{
  "title": "Too Many Requests",
  "detail": "Too Many Requests",
  "type": "about:blank",
  "status": 429
}

import stripe
import time
 
stripe.api_key = "sk_live_..."
 
def create_charge_with_retry(amount: int, customer_id: str) -> stripe.Charge:
    max_retries = 3
 
    for attempt in range(max_retries):
        try:
            return stripe.Charge.create(
                amount=amount,
                currency="usd",
                customer=customer_id,
                idempotency_key=f"charge_{customer_id}_{amount}_{int(time.time() // 60)}"
                # idempotency key changes every minute - safe to retry within same minute
            )
        except stripe.error.RateLimitError as e:
            retry_after = int(e.headers.get("Retry-After", 60))
            if attempt < max_retries - 1:
                time.sleep(retry_after)
            else:
                raise

# Bad: Only set headers on 429
if not allowed:
    response.headers["X-RateLimit-Remaining"] = 0
 
# Good: Always set headers
response.headers["X-RateLimit-Limit"] = limit
response.headers["X-RateLimit-Remaining"] = remaining
response.headers["X-RateLimit-Reset"] = reset_at

# Less useful: Unix timestamp (client must calculate delta)
response.headers["X-RateLimit-Reset"] = reset_timestamp
 
# More useful for retry: seconds to wait
response.headers["Retry-After"] = str(seconds_until_reset)
# Include BOTH for maximum compatibility

import hashlib
 
def make_rate_limit_key(api_key: str) -> str:
    # Hash the API key: Redis key contains only a hash, not the key itself
    key_hash = hashlib.sha256(api_key.encode()).hexdigest()[:16]
    return f"rl:apikey:{key_hash}"

def get_warmup_limit(user_id: str, base_limit: int) -> int:
    user_age_days = get_user_age_days(user_id)
    if user_age_days < 1:   return int(base_limit * 0.2)
    if user_age_days < 7:   return int(base_limit * 0.5)
    if user_age_days < 30:  return int(base_limit * 0.8)
    return base_limit

# Bad: Tests depend on real time, are flaky
def test_rate_limit():
    limiter = RateLimiter(limit=5, window=60)
    for i in range(5):
        assert limiter.is_allowed("user")  # passes
    assert not limiter.is_allowed("user")  # passes
    time.sleep(61)  # SLOW! Don't do this in unit tests
    assert limiter.is_allowed("user")  # passes
 
# Good: Inject clock for deterministic tests
from unittest.mock import patch
 
def test_rate_limit_with_mock_time():
    limiter = RateLimiter(limit=5, window=60)
    with patch("time.time") as mock_time:
        mock_time.return_value = 1000.0
        for i in range(5):
            assert limiter.is_allowed("user")
        assert not limiter.is_allowed("user")
 
        mock_time.return_value = 1061.0  # Advance 61 seconds
        assert limiter.is_allowed("user")  # Window reset

DRY_RUN = os.getenv("RATE_LIMIT_DRY_RUN", "false") == "true"
 
def check_rate_limit(identifier, limit, window) -> bool:
    result = redis_limiter.is_allowed(identifier, limit, window)
    if not result["allowed"]:
        if DRY_RUN:
            logger.warning(f"[DRY RUN] Would have rejected {identifier} "
                          f"(count={result['count']}, limit={limit})")
            return True  # Allow anyway in dry run
        return False
    return True

# Pass user rate limit context in request headers between microservices
headers = {
    "X-User-Id": user_id,
    "X-Rate-Limit-Remaining": str(remaining),
    "X-Request-Cost": str(cost)
}
requests.get("http://service-b/internal/data", headers=headers)

# Check how Redis is storing a rate limit key
OBJECT ENCODING rl:fw:user123:28956
# Returns: "int" (very efficient) or "embstr" or "raw"

# WRONG: This is your proxy/load balancer's IP
client_ip = request.remote_addr  # "10.0.0.5" (internal proxy)
 
# CORRECT: Get the real client IP from forwarded headers
def get_real_ip(request) -> str:
    forwarded_for = request.headers.get("X-Forwarded-For")
    if forwarded_for:
        # "X-Forwarded-For: client, proxy1, proxy2" - take the first
        return forwarded_for.split(",")[0].strip()
    return request.headers.get("X-Real-IP", request.remote_addr)

# WRONG: Key lives forever
redis.incr(f"rl:{user_id}")
 
# CORRECT: Always set expiry
redis.incr(f"rl:{user_id}")
redis.expire(f"rl:{user_id}", window_seconds)
# Or use SET with EX option atomically

# SAFER: Use pipeline or Lua
pipe = redis.pipeline()
pipe.incr(key)
pipe.expire(key, window)
pipe.execute()

import hashlib
 
def get_jittered_window_start(user_id: str, window_seconds: int) -> int:
    # Deterministic but distributed: each user has a unique window offset
    user_hash = int(hashlib.sha256(user_id.encode()).hexdigest(), 16)
    offset = user_hash % window_seconds
    now = int(time.time())
    window_base = now - (now % window_seconds)
    return window_base + offset - window_seconds  # previous window start

# WRONG: Naive sleep loop
while True:
    if not rate_limiter.is_allowed(user_id):
        time.sleep(1)  # Sleep 1 second, then try again
        continue
    # do work

# WRONG: Non-atomic check-then-increment
count = int(redis.get(key) or 0)
if count < limit:
    redis.set(key, count + 1)  # Race condition here
    return True
return False

# WRONG: One counter for the entire system
if redis.incr("global:counter") > 10000:
    return 429

# WRONG: Unbounded log grows forever
user_log = all_logs[user_id]
user_log.append(time.time())  # Never evict old entries

# WRONG: Ignore Retry-After, use fixed delay
except RateLimitError:
    time.sleep(5)  # Always wait 5 seconds, ignoring Retry-After
    retry()

# WRONG: Internal network requests skip rate limiting
if request.remote_addr.startswith("10."):
    return  # Skip rate limit for "internal" requests

# WRONG: Immediate retry loop
for attempt in range(100):
    response = api.call()
    if response.status_code == 429:
        continue  # Retry immediately - amplifies the problem!

def login(username, password):
    user = authenticate(username, password)
    if user:
        # Only rate limit successful logins?
        check_rate_limit(user.id)
        return generate_token(user)
    return 401

# WRONG: Drop request silently, return nothing or close connection
if rate_limited:
    return  # No response, connection just closes

# WRONG: Same limit for all endpoints regardless of cost
@rate_limit(limit=100, window=60)  # 100/min applied to everything
def all_endpoints():
    ...

# Metrics to expose (Prometheus format)
rate_limit_requests_total{endpoint, user_tier, result}  # allowed vs denied
rate_limit_utilization{user_id, tier}                   # % of limit used
rate_limit_denied_rate                                  # 429s per second
rate_limit_redis_latency_p99                           # Redis round trip
rate_limit_top_consumers                               # Top 10 heaviest users

# Prometheus alerting rules
groups:
  - name: rate_limiting
    rules:
      - alert: HighDenyRate
        expr: rate(rate_limit_denied_total[5m]) > 100
        for: 2m
        annotations:
          summary: "More than 100 rate limit denials per second for 2 minutes"
 
      - alert: RedisRateLimitDown
        expr: up{job="redis-rate-limit"} == 0
        for: 30s
        annotations:
          summary: "Redis rate limiter is down - fail-open mode active"
 
      - alert: SuspiciousHighUser
        expr: rate_limit_utilization > 0.95
        for: 5m
        annotations:
          summary: "User consistently near rate limit - potential abuse"

import structlog
 
logger = structlog.get_logger()
 
def log_rate_limit_event(
    identifier: str,
    endpoint: str,
    allowed: bool,
    count: int,
    limit: int
) -> None:
    log_fn = logger.info if allowed else logger.warning
    log_fn(
        "rate_limit_check",
        identifier=identifier,
        endpoint=endpoint,
        allowed=allowed,
        count=count,
        limit=limit,
        utilization=round(count / limit, 3),
        event_type="rate_limit_denied" if not allowed else "rate_limit_allowed"
    )

import pytest
from unittest.mock import patch, MagicMock
import time
 
 
class TestRateLimiter:
 
    def test_allows_requests_within_limit(self):
        limiter = RateLimiter(limit=5, window=60)
        for i in range(5):
            assert limiter.is_allowed("user1"), f"Request {i+1} should be allowed"
 
    def test_denies_requests_over_limit(self):
        limiter = RateLimiter(limit=5, window=60)
        for _ in range(5):
            limiter.is_allowed("user1")
        assert not limiter.is_allowed("user1"), "6th request should be denied"
 
    def test_different_users_have_independent_limits(self):
        limiter = RateLimiter(limit=5, window=60)
        for _ in range(5):
            limiter.is_allowed("user1")  # Exhaust user1's limit
        assert limiter.is_allowed("user2"), "User2 should not be affected"
 
    @patch("time.time")
    def test_window_resets_after_expiry(self, mock_time):
        mock_time.return_value = 1000.0
        limiter = RateLimiter(limit=5, window=60)
        for _ in range(5):
            limiter.is_allowed("user1")
        assert not limiter.is_allowed("user1")
 
        mock_time.return_value = 1061.0  # Advance past window
        assert limiter.is_allowed("user1"), "Should allow after window reset"
 
    def test_returns_correct_headers(self):
        limiter = RateLimiter(limit=5, window=60)
        result = limiter.check("user1")
        assert result["limit"] <mark class="obsidian-highlight"> 5
        assert result["remaining"] </mark> 4
        assert "reset_at" in result
 
    def test_concurrent_requests_respect_limit(self):
        import threading
        limiter = RateLimiter(limit=10, window=60)
        allowed_count = [0]
        lock = threading.Lock()
 
        def make_request():
            if limiter.is_allowed("user1"):
                with lock:
                    allowed_count[0] += 1
 
        threads = [threading.Thread(target=make_request) for _ in range(20)]
        for t in threads: t.start()
        for t in threads: t.join()
 
        assert allowed_count[0] <= 10, f"Expected <= 10 allowed, got {allowed_count[0]}"

import pytest
import redis as redis_lib
import fakeredis  # pip install fakeredis
 
 
@pytest.fixture
def fake_redis():
    """Use fakeredis for fast, isolated tests without a real Redis instance."""
    return fakeredis.FakeRedis(decode_responses=True)
 
 
@pytest.fixture
def redis_limiter(fake_redis):
    return RedisRateLimiter(
        redis_client=fake_redis,
        limit=10,
        window_seconds=60
    )
 
 
class TestRedisRateLimiter:
 
    def test_lua_script_atomicity(self, redis_limiter):
        """Verify Lua script prevents race conditions."""
        import concurrent.futures
 
        results = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
            futures = [
                executor.submit(redis_limiter.is_allowed, "user1")
                for _ in range(20)
            ]
            results = [f.result()["allowed"] for f in futures]
 
        allowed = sum(1 for r in results if r)
        assert allowed <= 10, f"Expected at most 10 allowed, got {allowed}"

# Using wrk to load test rate limiting behavior
wrk -t 10 -c 100 -d 30s \
    -H "X-API-Key: test_key_123" \
    http://localhost:8080/api/data
 
# Expected result: first N requests succeed (200), rest return 429
# Check rate of 429s in output