better exception handling

Author: igorbenav

src/app/api/dependencies.py

@@ -11,7 +11,7 @@
     Request
 )
 
-from app.api.exceptions import credentials_exception, privileges_exception
+from app.core.exceptions.http_exceptions import UnauthorizedException, ForbiddenException, RateLimitException
 from app.core.db.database import async_get_db
 from app.core.logger import logging
 from app.core.schemas import TokenData
@@ -28,38 +28,13 @@
 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)]
-) -> dict:
-    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
-    
-    if "@" in username_or_email:
-        user = await crud_users.get(db=db, email=token_data.username_or_email)
-    else: 
-        user = await crud_users.get(db=db, username=token_data.username_or_email)
-    
-    if user and not user["is_deleted"]:
-        return user
-    
-    raise credentials_exception
-
-
 async def get_current_user(
     token: Annotated[str, Depends(oauth2_scheme)],
     db: Annotated[AsyncSession, Depends(async_get_db)]
 ) -> dict:
     token_data = await verify_token(token, db)
     if token_data is None:
-        raise credentials_exception
+        raise UnauthorizedException("User not authenticated.")
 
     if "@" in token_data.username_or_email:
         user = await crud_users.get(db=db, email=token_data.username_or_email, is_deleted=False)
@@ -69,7 +44,7 @@ async def get_current_user(
     if user:
         return user
 
-    raise credentials_exception
+    raise UnauthorizedException("User not authenticated.")
 
 
 async def get_optional_user(
@@ -103,7 +78,7 @@ async def get_optional_user(
 
 async def get_current_superuser(current_user: Annotated[User, Depends(get_current_user)]) -> dict:
     if not current_user["is_superuser"]:
-        raise privileges_exception
+        raise ForbiddenException("You do not have enough privileges.")
 
     return current_user
 
@@ -143,7 +118,4 @@ async def rate_limiter(
         period=period
     )
     if is_limited:
-        raise HTTPException(
-            status_code=429,
-            detail="Rate limit exceeded"
-        )
+        raise RateLimitException("Rate limit exceeded.")

src/app/api/exceptions.py

@@ -9,7 +9,7 @@
 from app.core.db.database import async_get_db
 from app.core.schemas import Token
 from app.core.security import ACCESS_TOKEN_EXPIRE_MINUTES, create_access_token, authenticate_user
-from app.api.exceptions import credentials_exception
+from app.core.exceptions.http_exceptions import UnauthorizedException
 
 router = fastapi.APIRouter(tags=["login"])
 
@@ -24,7 +24,7 @@ async def login_for_access_token(
         db=db
     )
     if not user:
-        raise credentials_exception
+        raise UnauthorizedException("Wrong username, email or password.")
 
     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     access_token = await create_access_token(

src/app/api/v1/login.py

@@ -1,13 +1,14 @@
 from datetime import datetime
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, status
 from sqlalchemy.ext.asyncio import AsyncSession
 from jose import jwt, JWTError
 
 from app.core.security import oauth2_scheme, SECRET_KEY, ALGORITHM
 from app.core.db.database import async_get_db
 from app.core.db.crud_token_blacklist import crud_token_blacklist
 from app.core.schemas import TokenBlacklistCreate
+from app.core.exceptions.http_exceptions import UnauthorizedException
 
 router = APIRouter(tags=["login"])
 
@@ -28,8 +29,4 @@ async def logout(
         return {"message": "Logged out successfully"}
 
     except JWTError:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid token",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
+        raise UnauthorizedException("Invalid token.")

src/app/api/v1/logout.py

@@ -1,6 +1,6 @@
 from typing import Annotated
 
-from fastapi import Request, Depends, HTTPException
+from fastapi import Request, Depends
 from sqlalchemy.ext.asyncio import AsyncSession
 import fastapi
 
@@ -10,7 +10,7 @@
 from app.core.db.database import async_get_db
 from app.crud.crud_posts import crud_posts
 from app.crud.crud_users import crud_users
-from app.api.exceptions import privileges_exception
+from app.core.exceptions.http_exceptions import NotFoundException, ForbiddenException
 from app.core.utils.cache import cache
 from app.api.paginated import PaginatedListResponse, paginated_response, compute_offset
 
@@ -26,10 +26,10 @@ async def write_post(
 ):
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if db_user is None:
-        raise HTTPException(status_code=404, detail="User not found")
+        raise NotFoundException("User not found")
 
     if current_user["id"] != db_user["id"]:
-        raise privileges_exception
+        raise ForbiddenException()
 
     post_internal_dict = post.model_dump()
     post_internal_dict["created_by_user_id"] = db_user["id"]
@@ -53,7 +53,7 @@ async def read_posts(
 ):
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if not db_user:
-        raise HTTPException(status_code=404, detail="User not found")
+        raise NotFoundException("User not found")
 
     posts_data = await crud_posts.get_multi(
         db=db,
@@ -81,11 +81,11 @@ async def read_post(
 ):
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if db_user is None:
-        raise HTTPException(status_code=404, detail="User not found")
+        raise NotFoundException("User not found")
 
     db_post = await crud_posts.get(db=db, schema_to_select=PostRead, id=id, created_by_user_id=db_user["id"], is_deleted=False)
     if db_post is None:
-        raise HTTPException(status_code=404, detail="Post not found")
+        raise NotFoundException("Post not found")
 
     return db_post
 
@@ -106,14 +106,14 @@ async def patch_post(
 ):
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if db_user is None:
-        raise HTTPException(status_code=404, detail="User not found")
+        raise NotFoundException("User not found")
 
     if current_user["id"] != db_user["id"]:
-        raise privileges_exception
+        raise ForbiddenException()
 
     db_post = await crud_posts.get(db=db, schema_to_select=PostRead, id=id, is_deleted=False)
     if db_post is None:
-        raise HTTPException(status_code=404, detail="Post not found")
+        raise NotFoundException("Post not found")
 
     await crud_posts.update(db=db, object=values, id=id)
     return {"message": "Post updated"}
@@ -134,14 +134,14 @@ async def erase_post(
 ):
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if db_user is None:
-        raise HTTPException(status_code=404, detail="User not found")
+        raise NotFoundException("User not found")
 
     if current_user["id"] != db_user["id"]:
-        raise privileges_exception
+        raise ForbiddenException()
 
     db_post = await crud_posts.get(db=db, schema_to_select=PostRead, id=id, is_deleted=False)
     if db_post is None:
-        raise HTTPException(status_code=404, detail="Post not found")
+        raise NotFoundException("Post not found")
 
     await crud_posts.delete(db=db, db_row=db_post, id=id)
 
@@ -162,11 +162,11 @@ async def erase_db_post(
 ):
     db_user = await crud_users.get(db=db, schema_to_select=UserRead, username=username, is_deleted=False)
     if db_user is None:
-        raise HTTPException(status_code=404, detail="User not found")
+        raise NotFoundException("User not found")
 
     db_post = await crud_posts.get(db=db, schema_to_select=PostRead, id=id, is_deleted=False)
     if db_post is None:
-        raise HTTPException(status_code=404, detail="Post not found")
+        raise NotFoundException("Post not found")
 
     await crud_posts.db_delete(db=db, id=id)
     return {"message": "Post deleted from the database"}

src/app/api/v1/posts.py

@@ -7,6 +7,7 @@
 from app.api.dependencies import get_current_superuser
 from app.api.paginated import PaginatedListResponse, paginated_response, compute_offset
 from app.core.db.database import async_get_db
+from app.core.exceptions.http_exceptions import NotFoundException, DuplicateValueException, RateLimitException
 from app.crud.crud_rate_limit import crud_rate_limits
 from app.crud.crud_tier import crud_tiers
 from app.schemas.rate_limit import (
@@ -27,14 +28,14 @@ async def write_rate_limit(
 ):
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if not db_tier:
-        raise HTTPException(status_code=404, detail="Tier not found")
+        raise NotFoundException("Tier not found")
 
     rate_limit_internal_dict = rate_limit.model_dump()
     rate_limit_internal_dict["tier_id"] = db_tier["id"]
 
     db_rate_limit = await crud_rate_limits.exists(db=db, name=rate_limit_internal_dict["name"])
     if db_rate_limit:
-        raise HTTPException(status_code=400, detail="Rate Limit Name not available")
+        raise DuplicateValueException("Rate Limit Name not available")
 
     rate_limit_internal = RateLimitCreateInternal(**rate_limit_internal_dict)
     return await crud_rate_limits.create(db=db, object=rate_limit_internal)
@@ -50,7 +51,7 @@ async def read_rate_limits(
 ):
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if not db_tier:
-        raise HTTPException(status_code=404, detail="Tier not found")
+        raise NotFoundException("Tier not found")
 
     rate_limits_data = await crud_rate_limits.get_multi(
         db=db,
@@ -76,7 +77,7 @@ async def read_rate_limit(
 ):
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if not db_tier:
-        raise HTTPException(status_code=404, detail="Tier not found")
+        raise NotFoundException("Tier not found")
 
     db_rate_limit = await crud_rate_limits.get(
         db=db, 
@@ -85,7 +86,7 @@ async def read_rate_limit(
         id=id
     )
     if db_rate_limit is None:
-        raise HTTPException(status_code=404, detail="Rate Limit not found")
+        raise NotFoundException("Rate Limit not found")
 
     return db_rate_limit
 
@@ -100,7 +101,7 @@ async def patch_rate_limit(
 ):
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if db_tier is None:
-        raise HTTPException(status_code=404, detail="Tier not found")
+        raise NotFoundException("Tier not found")
 
     db_rate_limit = await crud_rate_limits.get(
         db=db,
@@ -109,19 +110,19 @@ async def patch_rate_limit(
         id=id
     )
     if db_rate_limit is None:
-        raise HTTPException(status_code=404, detail="Rate Limit not found")
+        raise NotFoundException("Rate Limit not found")
 
     db_rate_limit_path = await crud_rate_limits.exists(
         db=db,
         tier_id=db_tier["id"],
         path=values.path
     )
     if db_rate_limit_path is not None:
-        raise HTTPException(status_code=404, detail="There is already a rate limit for this path")
+        raise DuplicateValueException("There is already a rate limit for this path")
 
     db_rate_limit_name = await crud_rate_limits.exists(db=db)
     if db_rate_limit_path is not None:
-        raise HTTPException(status_code=404, detail="There is already a rate limit with this name")
+        raise DuplicateValueException("There is already a rate limit with this name")
 
     await crud_rate_limits.update(db=db, object=values, id=db_rate_limit["id"])
     return {"message": "Rate Limit updated"}
@@ -136,7 +137,7 @@ async def erase_rate_limit(
 ):
     db_tier = await crud_tiers.get(db=db, name=tier_name)
     if not db_tier:
-        raise HTTPException(status_code=404, detail="Tier not found")
+        raise NotFoundException("Tier not found")
 
     db_rate_limit = await crud_rate_limits.get(
         db=db, 
@@ -145,7 +146,7 @@ async def erase_rate_limit(
         id=id
     )
     if db_rate_limit is None:
-        raise HTTPException(status_code=404, detail="Rate Limit not found")
+        raise RateLimitException("Rate Limit not found")
 
     await crud_rate_limits.delete(db=db, db_row=db_rate_limit, id=db_rate_limit["id"])
     return {"message": "Rate Limit deleted"}

src/app/api/v1/rate_limits.py

@@ -12,6 +12,7 @@
 )
 from app.api.dependencies import get_current_superuser
 from app.core.db.database import async_get_db
+from app.core.exceptions.http_exceptions import DuplicateValueException, NotFoundException
 from app.crud.crud_tier import crud_tiers
 from app.api.paginated import PaginatedListResponse, paginated_response, compute_offset
 
@@ -26,7 +27,7 @@ async def write_tier(
     tier_internal_dict = tier.model_dump()
     db_tier = await crud_tiers.exists(db=db, name=tier_internal_dict["name"])
     if db_tier:
-        raise HTTPException(status_code=400, detail="Tier Name not available")
+        raise DuplicateValueException("Tier Name not available")
 
     tier_internal = TierCreateInternal(**tier_internal_dict)
     return await crud_tiers.create(db=db, object=tier_internal)
@@ -61,7 +62,7 @@ async def read_tier(
 ):
     db_tier = await crud_tiers.get(db=db, schema_to_select=TierRead, name=name)
     if db_tier is None:
-        raise HTTPException(status_code=404, detail="Tier not found")
+        raise NotFoundException("Tier not found")
 
     return db_tier
 
@@ -75,7 +76,7 @@ async def patch_tier(
 ):
     db_tier = await crud_tiers.get(db=db, schema_to_select=TierRead, name=name)
     if db_tier is None:
-        raise HTTPException(status_code=404, detail="Tier not found")
+        raise NotFoundException("Tier not found")
 
     await crud_tiers.update(db=db, object=values, name=name)
     return {"message": "Tier updated"}
@@ -89,7 +90,7 @@ async def erase_tier(
 ):
     db_tier = await crud_tiers.get(db=db, schema_to_select=TierRead, name=name)
     if db_tier is None:
-        raise HTTPException(status_code=404, detail="Tier not found")
+        raise NotFoundException("Tier not found")
 
     await crud_tiers.delete(db=db, db_row=db_tier, name=name)
     return {"message": "Tier deleted"}