Fixed linting issues.

Author: arsh-anwr

backend/app/core/rate_limiter/key_strategy/header_key_strategy.py

@@ -1,4 +1,5 @@
 from starlette.requests import Request
+
 from app.core.rate_limiter.key_strategy.key_strategy import KeyStrategy
 
 
@@ -8,4 +9,4 @@ def __init__(self, header_name: str = "X-Client-ID"):
 
     def get_key(self, request: Request, route_path: str) -> str:
         value = request.headers.get(self.header_name, "unknown")
-        return f"header:{self.header_name}:{value}:{route_path}"
+        return f"header:{self.header_name}:{value}:{route_path}"

backend/app/core/rate_limiter/key_strategy/ip_key_strategy.py

@@ -1,9 +1,11 @@
-from app.core.rate_limiter.key_strategy.key_strategy import KeyStrategy
 from starlette.requests import Request
 
+from app.core.rate_limiter.key_strategy.key_strategy import KeyStrategy
+
+
 class IPKeyStrategy(KeyStrategy):
     """Generate rate limit key based on client IP address."""
 
     def get_key(self, request: Request, route_path: str) -> str:
         client_ip = request.client.host if request.client else "unknown"
-        return f"ip:{client_ip}:{route_path}"
+        return f"ip:{client_ip}:{route_path}"

backend/app/core/rate_limiter/key_strategy/key_strategy.py

@@ -1,6 +1,8 @@
 from abc import ABC, abstractmethod
+
 from starlette.requests import Request
 
+
 class KeyStrategy(ABC):
     """Base interface for rate-limit key generation."""
 

backend/app/core/rate_limiter/key_strategy/key_strategy_enum.py

@@ -1,5 +1,6 @@
 from enum import Enum
 
+
 class KeyStrategyName(str, Enum):
     IP = "ip"
     HEADER = "header"

backend/app/core/rate_limiter/rate_limiter.py

@@ -1,5 +1,7 @@
 import logging
-from fastapi import Request, HTTPException
+
+from fastapi import HTTPException, Request
+
 from app.core.rate_limiter.key_strategy import key_strategy_registry
 from app.core.rate_limiter.key_strategy.key_strategy_enum import KeyStrategyName
 
@@ -11,15 +13,16 @@ def __init__(self, limit: int, window_seconds: int, key_policy: KeyStrategyName
         self.window_seconds = window_seconds
         self.key_policy = key_policy
 
-    async def __call__(self, request: Request):
+    async def __call__(self, request: Request) -> None:
         rate_limiter = getattr(request.app.state, "rate_limiter", None)
 
         if rate_limiter is None:
-            return
+            return None
 
         # Create Key
         key_strategy = key_strategy_registry.get_key_strategy(self.key_policy)
-        key = key_strategy.get_key(request, request.scope.get('path'))
+        path: str = request.scope.get("path") or ""
+        key = key_strategy.get_key(request, path)
 
         allowed = True
         retry_after = None
@@ -40,4 +43,4 @@ async def __call__(self, request: Request):
                 status_code=429,
                 detail=f"Too Many Requests. Retry after {retry_after}s",
                 headers={"Retry-After": str(retry_after)}
-            )
+            )

backend/app/core/rate_limiter/rate_limiting_algorithm/base.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
+
 from abc import ABC, abstractmethod
-from typing import Tuple, Optional
+
 
 class BaseRateLimiter(ABC):
     """Interface for pluggable rate limiter strategies."""
 
     @abstractmethod
-    async def allow_request(self, key: str, limit: int, window_seconds: int, member_id: Optional[str] = None) -> Tuple[bool, Optional[int]]:
+    async def allow_request(self, key: str, limit: int, window_seconds: int, member_id: str | None = None) -> tuple[bool, int | None]:
         """
         Return (allowed: bool, retry_after_seconds: Optional[int]).
         If allowed True -> retry_after_seconds is None.

backend/app/core/rate_limiter/rate_limiting_algorithm/registry.py

@@ -1,9 +1,13 @@
-from typing import Optional
-from app.core.rate_limiter.rate_limiting_algorithm.base import BaseRateLimiter
-from app.core.rate_limiter.rate_limiting_algorithm.sliding_window import SlidingWindowRateLimiter
+
 import redis.asyncio as redis
 
-def get_rate_limiter(strategy: Optional[str], redis_url: Optional[str], fail_open: Optional[str]) -> Optional[BaseRateLimiter]:
+from app.core.rate_limiter.rate_limiting_algorithm.base import BaseRateLimiter
+from app.core.rate_limiter.rate_limiting_algorithm.sliding_window import (
+    SlidingWindowRateLimiter,
+)
+
+
+def get_rate_limiter(strategy: str | None, redis_url: str | None, fail_open: bool | None) -> BaseRateLimiter | None:
     """
     Factory: returns an instance of BaseRateLimiter or None (if disabled).
     """
@@ -13,10 +17,10 @@ def get_rate_limiter(strategy: Optional[str], redis_url: Optional[str], fail_ope
     if not redis_url:
         return None
 
-    rc = redis.from_url(redis_url, encoding="utf-8", decode_responses=True)
+    rc: redis.Redis = redis.from_url(redis_url, encoding="utf-8", decode_responses=True) # type: ignore[no-untyped-call]
     st = strategy.lower()
     if st == "sliding_window" or st == "sliding-window":
         return SlidingWindowRateLimiter(rc, fail_open or False)
 
     # extendable for other strategies
-    raise ValueError(f"Unknown rate limiter strategy: {strategy}")
+    raise ValueError(f"Unknown rate limiter strategy: {strategy}")

backend/app/core/rate_limiter/rate_limiting_algorithm/sliding_window.py

@@ -1,15 +1,20 @@
-import time
 import logging
+import time
 from pathlib import Path
-from typing import Optional, Tuple
 
 import redis.asyncio as redis
 
 from app.core.rate_limiter.rate_limiting_algorithm.base import BaseRateLimiter
 
 logger = logging.getLogger(__name__)
 
-SCRIPT_PATH = Path(__file__).parent /".."/".."/".."/"alembic"/ "rate_limiting_algorithms" / "sliding_window.lua"
+SCRIPT_PATH = (
+    Path(__file__).resolve()
+    .parents[3]
+    / "alembic"
+    / "rate_limiting_algorithms"
+    / "sliding_window.lua"
+)
 
 
 class SlidingWindowRateLimiter(BaseRateLimiter):
@@ -18,7 +23,7 @@ def __init__(self, redis_client: redis.Redis, fail_open: bool):
         self.lua_script = None
         self.fail_open = fail_open
 
-    async def load_script(self):
+    async def load_script(self) -> str | None:
         if self.lua_script is None:
             script_text = SCRIPT_PATH.read_text()
             # LOAD script into redis → returns SHA
@@ -30,16 +35,18 @@ async def allow_request(
         key: str,
         limit: int,
         window_seconds: int,
-        member_id: Optional[str] = None
-    ) -> Tuple[bool, Optional[int]]:
+        member_id: str | None = None
+    ) -> tuple[bool, int | None]:
 
         now_ms = int(time.time() * 1000)
         window_ms = window_seconds * 1000
         member = member_id or f"{now_ms}"
 
         try:
             sha = await self.load_script()
-            res = await self.redis.evalsha(
+            if sha is None:
+                raise Exception
+            res = await self.redis.evalsha(# type: ignore[misc]
                 sha,
                 1,
                 key,
@@ -62,5 +69,5 @@ async def allow_request(
 
         return False, retry_after_s
 
-    def get_fail_open(self):
+    def get_fail_open(self) -> bool:
         return self.fail_open

backend/app/main.py

@@ -5,7 +5,6 @@
 
 from app.api.main import api_router
 from app.core.config import settings
-
 from app.core.rate_limiter.rate_limiting_algorithm.registry import get_rate_limiter
 
 

backend/tests/rate_limiter/test_dependency_rate_limit.py

@@ -1,7 +1,8 @@
+from unittest.mock import AsyncMock
+
 import pytest
-from fastapi import FastAPI, Depends
+from fastapi import Depends, FastAPI
 from fastapi.testclient import TestClient
-from unittest.mock import AsyncMock
 
 from app.core.rate_limiter.rate_limiter import RateLimiter