Better exceptions

README.md

@@ -430,7 +430,6 @@ First, you may want to take a look at the project structure and understand what
     │   ├── api                       # Folder containing API-related logic.
     │   │   ├── __init__.py
     │   │   ├── dependencies.py       # Defines dependencies for use across API endpoints.
-    │   │   ├── exceptions.py         # Custom exceptions for the API.
     │   │   ├── paginated.py          # Utilities for API response pagination.
     │   │   │
     │   │   └── v1                    # Version 1 of the API.
@@ -460,7 +459,8 @@ First, you may want to take a look at the project structure and understand what
     │   │   │
     │   │   ├── exceptions            # Custom exception classes.
     │   │   │   ├── __init__.py
-    │   │   │   └── exceptions.py     # Definitions of custom exceptions.
+    │   │   │   ├── cache_exceptions.py   # Exceptions related to cache operations.
+    │   │   │   └── http_exceptions.py    # HTTP-related exceptions.
     │   │   │
     │   │   └── utils                 # Utility functions and helpers.
     │   │       ├── __init__.py
@@ -890,6 +890,33 @@ async def read_entities(
     )
 ```
 
+#### 5.7.2 HTTP Exceptions
+
+To add exceptions you may just import from `app/core/exceptions/http_exceptions` and optionally add a detail:
+
+```python
+from app.core.exceptions.http_exceptions import NotFoundException
+
+# If you want to specify the detail, just add the message
+if not user:
+  raise NotFoundException("User not found")
+
+# Or you may just use the default message
+if not post:
+  raise NotFoundException()
+```
+
+**The predefined possibilities in http_exceptions are the following:**
+- `CustomException`: 500 internal error
+- `BadRequestException`: 400 bad request
+- `NotFoundException`: 404 not found
+- `ForbiddenException`: 403 forbidden
+- `UnauthorizedException`: 401 unauthorized
+- `UnprocessableEntityException`: 422 unprocessable entity
+- `DuplicateValueException`: 422 unprocessable entity
+- `RateLimitException`: 429 too many requests
+
+
 ### 5.8 Caching
 The `cache` decorator allows you to cache the results of FastAPI endpoint functions, enhancing response times and reducing the load on your application by storing and retrieving data in a cache.
 

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/v1/login.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/logout.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/posts.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/rate_limits.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"}