rate limiter dependency ready and working. Cache decorator updated

Author: igorbenav

src/app/api/dependencies.py

@@ -1,32 +1,44 @@
 from typing import Annotated
 
 from app.core.security import SECRET_KEY, ALGORITHM, oauth2_scheme
-from app.core.config import RedisRateLimiterSettings, settings
+from app.core.config import settings
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from jose import JWTError, jwt
 from fastapi import (
     Depends, 
     HTTPException, 
-    Request,
-    status
+    Request
 )
 
 from app.core.database import async_get_db
 from app.core.models import TokenData
-# from app.core.rate_limit import is_rate_limited
+from app.core.rate_limit import is_rate_limited
+from app.core.logger import logging
 from app.models.user import User
 from app.api.exceptions import credentials_exception, privileges_exception
 from app.crud.crud_users import crud_users
 from app.crud.crud_tier import crud_tiers
+from app.crud.crud_rate_limit import crud_rate_limits
+from app.schemas.rate_limit import sanitize_path
 
-async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: Annotated[AsyncSession, Depends(async_get_db)]) -> User:
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_LIMIT = settings.DEFAULT_RATE_LIMIT_LIMIT
+DEFAULT_PERIOD = settings.DEFAULT_RATE_LIMIT_PERIOD
+
+async def get_current_user(
+    token: Annotated[str, Depends(oauth2_scheme)],
+    db: Annotated[AsyncSession, Depends(async_get_db)]
+) -> User | None:
     try:
         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
         username_or_email: str = payload.get("sub")
         if username_or_email is None:
             raise credentials_exception
         token_data = TokenData(username_or_email=username_or_email)
+    
     except JWTError:
         raise credentials_exception
 
@@ -35,16 +47,80 @@ async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: An
     else: 
         user = await crud_users.get(db=db, username=token_data.username_or_email)
 
-    if user is None:
-        raise credentials_exception
+    if user and not user["is_deleted"]:
+        return user
 
-    if user.is_deleted:
-        raise HTTPException(status_code=400, detail="User deleted")
+    raise credentials_exception
+
+
+async def get_optional_user(
+    request: Request,
+    db: AsyncSession = Depends(async_get_db)
+) -> User | None:
+    token = request.headers.get("Authorization")
+    if not token:
+        return None
+
+    try:
+        token_type, _, token_value = token.partition(' ')
+        if token_type.lower() != 'bearer' or not token_value:
+            return None
+
+        return await get_current_user(token_value, db)
+    
+    except HTTPException as http_exc:
+        if http_exc.status_code != 401:
+            logger.error(f"Unexpected HTTPException in get_optional_user: {http_exc.detail}")
+        return None
+    
+    except Exception as exc:
+        logger.error(f"Unexpected error in get_optional_user: {exc}")
+        return None    
 
-    return user
 
 async def get_current_superuser(current_user: Annotated[User, Depends(get_current_user)]) -> User:
-    if not current_user.is_superuser:
+    if not current_user["is_superuser"]:
         raise privileges_exception
 
     return current_user
+
+
+async def rate_limiter(
+    request: Request,
+    db: Annotated[AsyncSession, Depends(async_get_db)],
+    user: User | None = Depends(get_optional_user)
+):
+    path = sanitize_path(request.url.path)
+    if user:
+        user_id = user["id"]
+        tier = await crud_tiers.get(db, id=user["tier_id"])
+        if tier:
+            rate_limit = await crud_rate_limits.get(
+                db=db,
+                tier_id=tier["id"],
+                path=path
+            )
+            if rate_limit:
+                limit, period = rate_limit["limit"], rate_limit["period"]
+            else:
+                logger.warning(f"User {user_id} with tier '{tier['name']}' has no specific rate limit for path '{path}'. Applying default rate limit.")
+                limit, period = DEFAULT_LIMIT, DEFAULT_PERIOD
+        else:
+            logger.warning(f"User {user_id} has no assigned tier. Applying default rate limit.")
+            limit, period = DEFAULT_LIMIT, DEFAULT_PERIOD
+    else:
+        user_id = request.client.host
+        limit, period = DEFAULT_LIMIT, DEFAULT_PERIOD
+
+    is_limited = await is_rate_limited(
+        db=db,
+        user_id=user_id,
+        path=path,
+        limit=limit,
+        period=period
+    )
+    if is_limited:
+        raise HTTPException(
+            status_code=429,
+            detail="Rate limit exceeded"
+        )

src/app/api/v1/login.py

@@ -29,7 +29,7 @@ async def login_for_access_token(
 
     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     access_token = await create_access_token(
-        data={"sub": user.username}, expires_delta=access_token_expires
+        data={"sub": user["username"]}, expires_delta=access_token_expires
     )
 
     return {"access_token": access_token, "token_type": "bearer"}

src/app/api/v1/posts.py

@@ -1,4 +1,4 @@
-from typing import List, Annotated
+from typing import Annotated
 
 from fastapi import Request, Depends, HTTPException
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -28,7 +28,7 @@ async def write_post(
     if db_user is None:
         raise HTTPException(status_code=404, detail="User not found")
 
-    if current_user.id != db_user["id"]:
+    if current_user["id"] != db_user["id"]:
         raise privileges_exception
 
     post_internal_dict = post.model_dump()
@@ -94,7 +94,7 @@ async def read_post(
 @cache(
     "{username}_post_cache", 
     resource_id_name="id", 
-    to_invalidate_extra={"{username}_posts": "{username}"}
+    pattern_to_invalidate_extra=["{username}_posts:*"]
 )
 async def patch_post(
     request: Request,
@@ -108,7 +108,7 @@ async def patch_post(
     if db_user is None:
         raise HTTPException(status_code=404, detail="User not found")
 
-    if current_user.id != db_user["id"]:
+    if current_user["id"] != db_user["id"]:
         raise privileges_exception
 
     db_post = await crud_posts.get(db=db, schema_to_select=PostRead, id=id, is_deleted=False)
@@ -136,7 +136,7 @@ async def erase_post(
     if db_user is None:
         raise HTTPException(status_code=404, detail="User not found")
 
-    if current_user.id != db_user.id:
+    if current_user["id"] != db_user["id"]:
         raise privileges_exception
 
     db_post = await crud_posts.get(db=db, schema_to_select=PostRead, id=id, is_deleted=False)

src/app/api/v1/rate_limits.py

@@ -4,28 +4,28 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 import fastapi
 
+from app.api.dependencies import get_current_superuser
+from app.api.paginated import PaginatedListResponse, paginated_response, compute_offset
+from app.core.database import async_get_db
+from app.crud.crud_rate_limit import crud_rate_limits
+from app.crud.crud_tier import crud_tiers
 from app.schemas.rate_limit import (
     RateLimitRead,
     RateLimitCreate,
     RateLimitCreateInternal,
     RateLimitUpdate
 )
-from app.api.dependencies import get_current_superuser
-from app.core.database import async_get_db
-from app.crud.crud_rate_limit import crud_rate_limits
-from app.crud.crud_tier import crud_tiers
-from app.api.paginated import PaginatedListResponse, paginated_response, compute_offset
 
 router = fastapi.APIRouter(tags=["rate_limits"])
 
-@router.post("/tier/{name}/rate_limit", dependencies=[Depends(get_current_superuser)], status_code=201)
+@router.post("/tier/{tier_name}/rate_limit", dependencies=[Depends(get_current_superuser)], status_code=201)
 async def write_rate_limit(
     request: Request, 
-    name: str,
+    tier_name: str,
     rate_limit: RateLimitCreate, 
     db: Annotated[AsyncSession, Depends(async_get_db)]
 ):
-    db_tier = await crud_tiers.get(db=db, name=name)
+    db_tier = await crud_tiers.get(db=db, name=tier_name)
     if not db_tier:
         raise HTTPException(status_code=404, detail="Tier not found")
 
@@ -40,15 +40,15 @@ async def write_rate_limit(
     return await crud_rate_limits.create(db=db, object=rate_limit_internal)
 
 
-@router.get("/tier/{name}/rate_limits", response_model=PaginatedListResponse[RateLimitRead])
+@router.get("/tier/{tier_name}/rate_limits", response_model=PaginatedListResponse[RateLimitRead])
 async def read_rate_limits(
     request: Request,
-    name: str,
+    tier_name: str,
     db: Annotated[AsyncSession, Depends(async_get_db)],
     page: int = 1,
     items_per_page: int = 10
 ):
-    db_tier = await crud_tiers.get(db=db, name=name)
+    db_tier = await crud_tiers.get(db=db, name=tier_name)
     if not db_tier:
         raise HTTPException(status_code=404, detail="Tier not found")
 
@@ -90,7 +90,7 @@ async def read_rate_limit(
     return db_rate_limit
 
 
-@router.patch("/tier/{tier_name}/rate_limit/{path}", dependencies=[Depends(get_current_superuser)])
+@router.patch("/tier/{tier_name}/rate_limit/{id}", dependencies=[Depends(get_current_superuser)])
 async def patch_rate_limit(
     request: Request,
     tier_name: str,
@@ -99,23 +99,31 @@ async def patch_rate_limit(
     db: Annotated[AsyncSession, Depends(async_get_db)]
 ):
     db_tier = await crud_tiers.get(db=db, name=tier_name)
-    if not db_tier:
+    if db_tier is None:
         raise HTTPException(status_code=404, detail="Tier not found")
 
     db_rate_limit = await crud_rate_limits.get(
-        db=db, 
+        db=db,
         schema_to_select=RateLimitRead, 
         tier_id=db_tier["id"],
         id=id
     )
     if db_rate_limit is None:
         raise HTTPException(status_code=404, detail="Rate Limit not found")
 
+    db_rate_limit_path = await crud_rate_limits.exists(
+        db=db,
+        tier_id=db_tier["id"],
+        path=values.path
+    )
+
+    db_rate_limit_name = await crud_rate_limits.exists(db=db)
+
     await crud_rate_limits.update(db=db, object=values, id=db_rate_limit["id"])
     return {"message": "Rate Limit updated"}
 
 
-@router.delete("/tier/{tier_name}/rate_limit/{path}", dependencies=[Depends(get_current_superuser)])
+@router.delete("/tier/{tier_name}/rate_limit/{id}", dependencies=[Depends(get_current_superuser)])
 async def erase_rate_limit(
     request: Request,
     tier_name: str,

src/app/api/v1/tasks.py

@@ -1,12 +1,13 @@
 from arq.jobs import Job as ArqJob
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, Depends
 
 from app.core import queue
 from app.schemas.job import Job
+from app.api.dependencies import rate_limiter
 
 router = APIRouter(prefix="/tasks", tags=["tasks"])
 
-@router.post("/task", response_model=Job, status_code=201)
+@router.post("/task", response_model=Job, status_code=201, dependencies=[Depends(rate_limiter)])
 async def create_task(message: str):
     job = await queue.pool.enqueue_job("sample_background_task", message)
     return {"id": job.job_id}

src/app/api/v1/tiers.py

@@ -1,4 +1,4 @@
-from typing import List, Annotated
+from typing import Annotated
 
 from fastapi import Request, Depends, HTTPException
 from sqlalchemy.ext.asyncio import AsyncSession