Auto-Generated API Reference¶

def check(self, key: str, limit: int, window_seconds: int) -> RateLimitResult:
    """Check rate limit for a key and return the remaining quota."""
    raise NotImplementedError

def check(self, key: str, limit: int, window_seconds: int) -> RateLimitResult:
    """Check rate limit for key and return the result."""
    now = int(time.time())
    if now - self._last_sweep >= self._sweep_interval_seconds:
        self._sweep(now)
    elif len(self._state) >= self._max_entries:
        self._sweep(now)
    state = self._state.get(key)
    if state is None or state.reset_at <= now:
        reset_at = now + window_seconds
        self._state[key] = RateLimitState(count=1, reset_at=reset_at)
        remaining = max(limit - 1, 0)
        return RateLimitResult(allowed=True, remaining=remaining, reset_at=reset_at)

    state.count += 1
    allowed = state.count <= limit
    remaining = max(limit - state.count, 0)
    return RateLimitResult(allowed=allowed, remaining=remaining, reset_at=state.reset_at)

def check(self, key: str, limit: int, window_seconds: int) -> RateLimitResult:
    """Check rate limit for key against Redis and return the result."""
    now = int(time.time())
    redis_key = self._key(key)
    script = """
    local current = redis.call("INCR", KEYS[1])
    if current == 1 then
      redis.call("EXPIRE", KEYS[1], ARGV[1])
    end
    local ttl = redis.call("TTL", KEYS[1])
    return {current, ttl}
    """
    redis_client = cast(Any, self._redis)
    result = redis_client.eval(script, 1, redis_key, window_seconds)
    count, ttl = result
    if ttl is None or int(ttl) < 0:
        ttl = window_seconds
    reset_at = now + int(ttl)
    allowed = int(count) <= limit
    remaining = max(limit - int(count), 0)
    return RateLimitResult(allowed=allowed, remaining=remaining, reset_at=reset_at)

def is_blocked(self, key: str) -> tuple[bool, int]:
    """Return (blocked, retry_after_seconds)."""
    raise NotImplementedError

def record_failure(self, key: str) -> tuple[bool, int]:
    """Record a failed attempt and return (blocked, retry_after_seconds)."""
    raise NotImplementedError

def reset(self, key: str) -> None:
    """Clear rate limit state for a key."""
    raise NotImplementedError

def remaining(self, now: float) -> int:
    """Return remaining seconds until window expiry."""
    return max(0, int(self.expires_at - now))

def is_blocked(self, key: str) -> tuple[bool, int]:
    """Return whether the key is currently blocked and retry-after seconds."""
    now = time.time()
    state = self._get_state(key, now)
    if state is None:
        return False, 0
    if state.count >= self.max_attempts:
        return True, state.remaining(now)
    return False, 0

def record_failure(self, key: str) -> tuple[bool, int]:
    """Record a failed attempt and return blocked status and retry-after seconds."""
    now = time.time()
    state = self._get_state(key, now)
    if state is None:
        state = RateLimitState(count=1, expires_at=now + self.window_seconds)
        self._state[key] = state
    else:
        state.count += 1
    blocked = state.count >= self.max_attempts
    return blocked, state.remaining(now)

def reset(self, key: str) -> None:
    """Clear rate limit state for the given key."""
    self._state.pop(key, None)

def is_blocked(self, key: str) -> tuple[bool, int]:
    """Return whether the key is currently blocked and retry-after seconds."""
    redis_key = self._key(key)
    value = cast(Any, self.redis.get(redis_key))
    if value is None:
        return False, 0
    try:
        count = int(value)
    except ValueError:
        self.redis.delete(redis_key)
        return False, 0
    if count >= self.max_attempts:
        return True, self._ttl(redis_key)
    return False, 0

def record_failure(self, key: str) -> tuple[bool, int]:
    """Record a failed attempt and return blocked status and retry-after seconds."""
    redis_key = self._key(key)
    pipe = self.redis.pipeline()
    pipe.incr(redis_key)
    pipe.ttl(redis_key)
    count, ttl = pipe.execute()
    if ttl is None or int(ttl) < 0:
        self.redis.expire(redis_key, self.window_seconds)
        ttl = self.window_seconds
    blocked = int(count) >= self.max_attempts
    return blocked, int(ttl)

def reset(self, key: str) -> None:
    """Clear rate limit state for the given key."""
    self.redis.delete(self._key(key))

def store_refresh(self, jti: str, user_id: str, ttl_seconds: int) -> None:
    """Store a refresh token identifier with TTL."""
    ...

def is_refresh_active(self, jti: str) -> bool:
    """Return True if the refresh token is active."""
    ...

def revoke_refresh(self, jti: str) -> None:
    """Revoke a refresh token identifier."""
    ...

def revoke_access(self, jti: str, ttl_seconds: int) -> None:
    """Mark an access token as revoked for the remaining TTL."""
    ...

def is_access_revoked(self, jti: str) -> bool:
    """Return True if the access token has been revoked."""
    ...

def store_refresh(self, jti: str, user_id: str, ttl_seconds: int) -> None:
    """Store a refresh token with the given TTL."""
    self._refresh[jti] = time.time() + ttl_seconds

def is_refresh_active(self, jti: str) -> bool:
    """Return True if the refresh token is active."""
    return self._is_active(self._refresh, jti)

def revoke_refresh(self, jti: str) -> None:
    """Revoke a refresh token by JTI."""
    self._refresh.pop(jti, None)

def revoke_access(self, jti: str, ttl_seconds: int) -> None:
    """Mark an access token as revoked for the remaining TTL."""
    self._revoked[jti] = time.time() + ttl_seconds

def is_access_revoked(self, jti: str) -> bool:
    """Return True if the access token has been revoked."""
    return self._is_active(self._revoked, jti)

def store_refresh(self, jti: str, user_id: str, ttl_seconds: int) -> None:
    """Store a refresh token with the given TTL in Redis."""
    # Redis 8 deprecates setex(key, ttl, value); use set(..., ex=ttl).
    self.redis.set(self._refresh_key(jti), user_id, ex=ttl_seconds)

def is_refresh_active(self, jti: str) -> bool:
    """Return True if the refresh token is active in Redis."""
    return self.redis.exists(self._refresh_key(jti)) == 1

def revoke_refresh(self, jti: str) -> None:
    """Revoke a refresh token in Redis."""
    self.redis.delete(self._refresh_key(jti))

def revoke_access(self, jti: str, ttl_seconds: int) -> None:
    """Mark an access token as revoked in Redis."""
    # Redis 8 deprecates setex(key, ttl, value); use set(..., ex=ttl).
    self.redis.set(self._revoked_key(jti), "1", ex=ttl_seconds)

def is_access_revoked(self, jti: str) -> bool:
    """Return True if the access token has been revoked in Redis."""
    return self.redis.exists(self._revoked_key(jti)) == 1

def get(self, key: str) -> str | None:
    """Return cached value for *key*, or None when absent/expired."""

def set(self, key: str, value: str, *, ttl_seconds: int) -> None:
    """Store *value* for *key* with TTL."""

def delete(self, key: str) -> None:
    """Delete *key* if present."""

def close(self) -> None:
    """Release any backend resources."""

def get(self, key: str) -> str | None:
    """Return the cached value for key, or None if absent or expired."""
    with self._lock:
        entry = self._entries.get(key)
        if entry is None:
            return None
        if entry.expires_at < time.time():
            self._entries.pop(key, None)
            return None
        return entry.value

def set(self, key: str, value: str, *, ttl_seconds: int) -> None:
    """Store value for key with the given TTL."""
    expires_at = time.time() + max(1, ttl_seconds)
    with self._lock:
        self._entries[key] = _MemoryEntry(value=value, expires_at=expires_at)

def delete(self, key: str) -> None:
    """Delete key from the cache if present."""
    with self._lock:
        self._entries.pop(key, None)

def close(self) -> None:
    """Clear all cache entries."""
    with self._lock:
        self._entries.clear()

def get(self, key: str) -> str | None:
    """Return the cached value for key, or None on miss or error."""
    try:
        value = self._redis.get(key)
    except RedisError as exc:
        logger.warning("Redis cache get failed for {}: {}", key, exc)
        return None
    return value if isinstance(value, str) else None

def set(self, key: str, value: str, *, ttl_seconds: int) -> None:
    """Store value for key with the given TTL in Redis."""
    try:
        # Redis 8 deprecates setex(key, ttl, value); use set(..., ex=ttl).
        # Keep the >=1s clamp (Redis rejects a 0/negative expiry).
        self._redis.set(key, value, ex=max(1, ttl_seconds))
    except RedisError as exc:
        logger.warning("Redis cache set failed for {}: {}", key, exc)

def delete(self, key: str) -> None:
    """Delete key from Redis if present."""
    try:
        self._redis.delete(key)
    except RedisError as exc:
        logger.warning("Redis cache delete failed for {}: {}", key, exc)

def close(self) -> None:
    """Close the Redis connection."""
    try:
        self._redis.close()
    except RedisError as exc:
        logger.warning("Redis cache close failed: {}", exc)

async def dispatch(
    self,
    request: Request,
    call_next: Callable[[Request], Awaitable[Response]],
) -> Response:
    """Process an HTTP request through the rate limiter."""
    if not self._settings.rate_limit_enabled or request.scope.get("type") != "http":
        return await call_next(request)

    path = request.url.path
    if self._is_exempt(path):
        return await call_next(request)

    rule = self._rule_for_path(path)
    limit = self._settings.rate_limit_default_requests
    window = self._settings.rate_limit_default_window_seconds
    if rule:
        limit = rule.get("requests", limit)
        window = rule.get("window_seconds", window)

    key = f"{self._client_id(request)}:{path}"
    result = self._limiter.check(key, limit, window)
    if not result.allowed:
        retry_after = max(result.reset_at - int(time.time()), 0)
        response: Response = JSONResponse(
            status_code=429,
            content={
                "error": "rate_limited",
                "message": "Rate limit exceeded. Try again later.",
            },
        )
        response.headers["Retry-After"] = str(retry_after)
        self._apply_headers(response, result, limit)
        return response

    response = await call_next(request)
    self._apply_headers(response, result, limit)
    return response

async def dispatch(
    self,
    request: Request,
    call_next: Callable[[Request], Awaitable[Response]],
) -> Response:
    """Process an HTTP request and attach security headers to the response."""
    response = await call_next(request)
    if not self._settings.security_headers_enabled:
        return response

    response.headers.setdefault("X-Frame-Options", "DENY")
    response.headers.setdefault("X-Content-Type-Options", "nosniff")
    response.headers.setdefault("X-XSS-Protection", "1; mode=block")
    response.headers.setdefault("Referrer-Policy", self._settings.security_referrer_policy)
    response.headers.setdefault(
        "Permissions-Policy",
        "geolocation=(), microphone=(), camera=(), payment=()",
    )
    response.headers.setdefault("Content-Security-Policy", self._settings.security_csp)

    if request.url.scheme == "https" and self._settings.security_hsts_seconds > 0:
        hsts = f"max-age={self._settings.security_hsts_seconds}"
        if self._settings.security_hsts_subdomains:
            hsts += "; includeSubDomains"
        response.headers.setdefault("Strict-Transport-Security", hsts)

    return response