Added rate limiter for fastAPI template.

Author: arsh-anwr

.env

@@ -43,3 +43,10 @@ SENTRY_DSN=
 # Configure these with your own Docker registry images
 DOCKER_IMAGE_BACKEND=backend
 DOCKER_IMAGE_FRONTEND=frontend
+
+# Redis config
+REDIS_URL=redis://redis:6379/0
+
+# Rate Limiting config
+RATE_LIMITER_STRATEGY=sliding_window
+RATE_LIMIT_FAIL_OPEN=false

backend/app/alembic/rate_limiting_algorithms/sliding_window.lua

@@ -0,0 +1,34 @@
+-- KEYS[1] = key
+-- ARGV[1] = now_ms
+-- ARGV[2] = window_ms
+-- ARGV[3] = limit
+-- ARGV[4] = member
+
+local key = KEYS[1]
+local now = tonumber(ARGV[1])
+local window = tonumber(ARGV[2])
+local limit = tonumber(ARGV[3])
+local member = ARGV[4]
+
+local min_score = now - window
+
+-- remove old entries
+redis.call('ZREMRANGEBYSCORE', key, 0, min_score)
+
+-- add current
+redis.call('ZADD', key, now, member)
+
+-- count
+local cnt = redis.call('ZCARD', key)
+
+-- expire same as window
+redis.call('PEXPIRE', key, window)
+
+-- fetch oldest
+local earliest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
+local oldest_ts = 0
+if earliest ~= false and earliest ~= nil and #earliest >= 2 then
+  oldest_ts = earliest[2]
+end
+
+return {cnt, oldest_ts}

backend/app/api/routes/users.py

@@ -11,6 +11,8 @@
     get_current_active_superuser,
 )
 from app.core.config import settings
+from app.core.rate_limiter.key_strategy.key_strategy_enum import KeyStrategyName
+from app.core.rate_limiter.rate_limiter import RateLimiter
 from app.core.security import get_password_hash, verify_password
 from app.models import (
     Item,
@@ -31,7 +33,8 @@
 
 @router.get(
     "/",
-    dependencies=[Depends(get_current_active_superuser)],
+    dependencies=[Depends(RateLimiter(limit=10, window_seconds=60, key_policy=KeyStrategyName.IP)),
+    Depends(get_current_active_superuser)],
     response_model=UsersPublic,
 )
 def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:

backend/app/core/config.py

@@ -41,13 +41,32 @@ class Settings(BaseSettings):
         list[AnyUrl] | str, BeforeValidator(parse_cors)
     ] = []
 
+    # Correct Redis default inside Docker Compose
+    REDIS_URL: str = ""
+    RATE_LIMITER_STRATEGY: Literal[
+        "none", "sliding_window"
+    ] = "none"
+    RATE_LIMIT_FAIL_OPEN: bool = True
+
     @computed_field  # type: ignore[prop-decorator]
     @property
     def all_cors_origins(self) -> list[str]:
         return [str(origin).rstrip("/") for origin in self.BACKEND_CORS_ORIGINS] + [
             self.FRONTEND_HOST
         ]
 
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def rate_limit_enabled(self) -> bool:
+        """
+        Returns True if rate limiting should be enabled based on strategy.
+        Mirrors the style of all_cors_origins.
+        """
+        strategy = (self.RATE_LIMITER_STRATEGY or "").strip().lower()
+        redis_url = (self.REDIS_URL or "").strip()
+
+        return strategy not in ("", "none") and bool(redis_url)
+
     PROJECT_NAME: str
     SENTRY_DSN: HttpUrl | None = None
     POSTGRES_SERVER: str

backend/app/core/rate_limiter/__init__.py

@@ -0,0 +1,11 @@
+from starlette.requests import Request
+from app.core.rate_limiter.key_strategy.key_strategy import KeyStrategy
+
+
+class HeaderKeyStrategy(KeyStrategy):
+    def __init__(self, header_name: str = "X-Client-ID"):
+        self.header_name = header_name
+
+    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}"

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

@@ -0,0 +1,9 @@
+from app.core.rate_limiter.key_strategy.key_strategy import KeyStrategy
+from starlette.requests import Request
+
+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}"

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

@@ -0,0 +1,10 @@
+from abc import ABC, abstractmethod
+from starlette.requests import Request
+
+class KeyStrategy(ABC):
+    """Base interface for rate-limit key generation."""
+
+    @abstractmethod
+    def get_key(self, request: Request, route_path: str) -> str:
+        """Return unique identifier string (e.g., 'ip:127.0.0.1')"""
+        raise NotImplementedError

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

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