feat: update submodule and remove unusing functional

.gitmodules

@@ -1,4 +1,3 @@
 [submodule "src/common"]
 	path = src/common
-	url = https://github.com/flavien-hugs/backend-common-code.git
-    branch = develop
+	url = https://github.com/flavien-hugs/common.git

.isort.cfg

@@ -1,2 +1,2 @@
 [settings]
-known_third_party = beanie,email_validator,fastapi,fastapi_cache,fastapi_jwt,fastapi_pagination,httpx,jinja2,jose,mongomock_motor,motor,pwdlib,pydantic,pydantic_settings,pymongo,pyotp,pytest,pytest_asyncio,slugify,starlette,typer,uvicorn,yaml
+known_third_party = beanie,email_validator,fastapi,fastapi_cache,fastapi_jwt,fastapi_pagination,httpx,jinja2,jose,mongomock_motor,pwdlib,pydantic,pydantic_settings,pymongo,pyotp,pytest,pytest_asyncio,slugify,starlette,typer,uvicorn,yaml

src/__init__.py

@@ -14,11 +14,13 @@
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 
-from src.common.helpers.appdesc import load_app_description, load_permissions
+from src.common.config import shutdown_db_client, startup_db_client
+from src.common.config.setup_permission import load_app_description, load_app_permissions
 from src.common.helpers.caching import init_redis_cache
 from src.common.helpers.error_codes import AppErrorCode
-from src.common.helpers.exceptions import setup_exception_handlers
-from src.config import settings, shutdown_db, startup_db
+from src.common.helpers.exception import setup_exception_handlers
+from src.common.middleware import RateLimitMiddleware
+from src.config import settings
 from src.models import Params, Role, User
 from src.routers import auth_router, param_router, perm_router, role_router, user_router
 from src.services import roles, users
@@ -33,10 +35,12 @@ class State(TypedDict):
 
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncIterator[State]:
-    await startup_db(app=app, models=[User, Role, Params])
+    await startup_db_client(
+        app=app, mongodb_uri=settings.MONGODB_URI, database_name=settings.MONGO_DB, document_models=[User, Role, Params]
+    )
 
     await load_app_description(mongodb_client=app.mongo_db_client)
-    await load_permissions(mongodb_client=app.mongo_db_client)
+    await load_app_permissions(mongodb_client=app.mongo_db_client)
 
     await roles.create_admin_role()
     await users.create_admin_user()
@@ -46,7 +50,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[State]:
     await init_redis_cache(app_name=BASE_URL, cache_db_url=settings.CACHE_DB_URL)
 
     yield
-    await shutdown_db(app=app)
+    await shutdown_db_client(app=app)
 
 
 app: FastAPI = FastAPI(
@@ -60,6 +64,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[State]:
 
 # Compress responses larger than 1000 bytes
 app.add_middleware(GZipMiddleware, minimum_size=int(settings.COMPRESS_MIN_SIZE))
+app.add_middleware(RateLimitMiddleware, limit=settings.RATE_LIMIT_REQUEST, interval=settings.RATE_LIMIT_INTERVAL)
 
 
 @app.get("/", include_in_schema=False)

src/config/__init__.py

@@ -1,4 +1,3 @@
-from .database import shutdown_db, startup_db  # noqa: F401
 from .email import get_email_settings  # noqa: F401
 from .settings import get_settings  # noqa: F401
 from .sms import get_sms_config  # noqa: F401
@@ -11,4 +10,4 @@
 email_settings = get_email_settings()
 enable_endpoint = enable_endpoint()
 
-__all__ = ["settings", "startup_db", "shutdown_db", "email_settings", "jwt_settings", "sms_config", "enable_endpoint"]
+__all__ = ["settings", "email_settings", "jwt_settings", "sms_config", "enable_endpoint"]

src/config/settings.py

@@ -48,7 +48,9 @@ class AuthBaseConfig(BaseSettings):
     EXPIRE_CACHE: Optional[PositiveInt] = Field(default=500, alias="EXPIRE_CACHE")
 
     # MIDDLEWARE CONFIG
-    COMPRESS_MIN_SIZE: Optional[PositiveInt] = Field(default=1_000, alias="COMPRESS_MIN_SIZE")
+    COMPRESS_MIN_SIZE: Optional[int] = Field(default=1000, alias="COMPRESS_MIN_SIZE")
+    RATE_LIMIT_REQUEST: Optional[int] = Field(default=5, alias="RATE_LIMIT_REQUEST")
+    RATE_LIMIT_INTERVAL: Optional[int] = Field(default=3600, alias="RATE_LIMIT_INTERVAL")
 
 
 @lru_cache

src/middleware/auth.py

@@ -12,7 +12,7 @@
 from slugify import slugify
 
 from src.common.helpers.caching import custom_key_builder as cache_key_builder
-from src.common.helpers.exceptions import CustomHTTException
+from src.common.helpers.exception import CustomHTTPException
 from src.config import jwt_settings, settings
 from src.services import users
 from src.shared import blacklist_token
@@ -71,7 +71,7 @@ def decode_access_token(cls, token: str) -> dict:
                 algorithms=[jwt_settings.JWT_ALGORITHM],
             )
         except (jwt.ExpiredSignatureError, JWTError) as err:
-            raise CustomHTTException(
+            raise CustomHTTPException(
                 code_error=AuthErrorCode.AUTH_INVALID_ACCESS_TOKEN,
                 message_error=str(err),
                 status_code=status.HTTP_401_UNAUTHORIZED,
@@ -86,14 +86,14 @@ async def verify_access_token(cls, token: str) -> bool:
 
         :param token: The access token to verify.
         :type token: str
-        :return: True if the token is valid, otherwise raises a CustomHTTException.
+        :return: True if the token is valid, otherwise raises a CustomHTTPException.
         :rtype: bool
-        :raises CustomHTTException: If the token is expired or invalid, raises a CustomHTTException.
+        :raises CustomHTTPException: If the token is expired or invalid, raises a CustomHTTPException.
         """
 
         try:
             if await blacklist_token.is_token_blacklisted(token):
-                raise CustomHTTException(
+                raise CustomHTTPException(
                     code_error=AuthErrorCode.AUTH_EXPIRED_ACCESS_TOKEN,
                     message_error="Token has expired !",
                     status_code=status.HTTP_401_UNAUTHORIZED,
@@ -105,13 +105,13 @@ async def verify_access_token(cls, token: str) -> bool:
             token_exp = decode_token.get("exp", 0)
             if check_if_active is True and token_exp > current_timestamp:
                 return True
-            raise CustomHTTException(
+            raise CustomHTTPException(
                 code_error=AuthErrorCode.AUTH_EXPIRED_ACCESS_TOKEN,
                 message_error="Token has expired !",
                 status_code=status.HTTP_401_UNAUTHORIZED,
             )
         except (ExpiredSignatureError, JWTError) as err:
-            raise CustomHTTException(
+            raise CustomHTTPException(
                 code_error=AuthErrorCode.AUTH_INVALID_ACCESS_TOKEN,
                 message_error=str(err),
                 status_code=status.HTTP_401_UNAUTHORIZED,
@@ -134,9 +134,9 @@ async def check_permissions(cls, token: str, required_permissions: set[str] = ()
         :type token: str
         :param required_permissions: A set of required permissions.
         :type required_permissions: set[str]
-        :return: True if the user has the required permissions, otherwise raises a CustomHTTException.
+        :return: True if the user has the required permissions, otherwise raises a CustomHTTPException.
         :rtype: bool
-        :raises CustomHTTException: If the user doesn't have the required permissions.
+        :raises CustomHTTPException: If the user doesn't have the required permissions.
         """
 
         docode_token = cls.decode_access_token(token)
@@ -151,7 +151,7 @@ async def check_permissions(cls, token: str, required_permissions: set[str] = ()
         if required_permissions & user_permissions:
             return True
         else:
-            raise CustomHTTException(
+            raise CustomHTTPException(
                 code_error=AuthErrorCode.AUTH_INSUFFICIENT_PERMISSION,
                 message_error="You do not have the necessary permissions to access this resource.",
                 status_code=status.HTTP_403_FORBIDDEN,
@@ -174,15 +174,15 @@ class AuthorizedHTTPBearer(HTTPBearer):
     async def __call__(self, request: Request):
         if auth := await super().__call__(request=request):
             if not (auth.scheme.lower() == "bearer" and auth.scheme.startswith("Bearer")):
-                raise CustomHTTException(
+                raise CustomHTTPException(
                     code_error=AuthErrorCode.AUTH_MISSING_SCHEME,
                     message_error="Missing or invalid authentication scheme.",
                     status_code=status.HTTP_401_UNAUTHORIZED,
                 )
             await CustomAccessBearer.verify_access_token(auth.credentials)
             return auth.credentials
 
-        raise CustomHTTException(
+        raise CustomHTTPException(
             code_error=AuthErrorCode.AUTH_EXPIRED_ACCESS_TOKEN,
             message_error="The token has expired.",
             status_code=status.HTTP_401_UNAUTHORIZED,
@@ -204,7 +204,7 @@ def __init__(self, required_permissions: set[str]):
 
     async def __call__(self, request: Request):
         if not (authorization := request.headers.get("Authorization")):
-            raise CustomHTTException(
+            raise CustomHTTPException(
                 code_error=AuthErrorCode.AUTH_MISSING_TOKEN,
                 message_error="Missing token.",
                 status_code=status.HTTP_401_UNAUTHORIZED,
@@ -214,7 +214,7 @@ async def __call__(self, request: Request):
 
 
 class CheckUserAccessHandler:
-    """ "
+    """
     Handler for checking user access based on the provided key and roles.
 
     This class provides a callable interface to check if a user has access to a resource
@@ -223,9 +223,9 @@ class CheckUserAccessHandler:
 
     :param key: The key to check for the resource.
     :rtype key: str
-    :return: True if the user has access to the resource, otherwise raises a CustomHTTException.
+    :return: True if the user has access to the resource, otherwise raises a CustomHTTPException.
     :rtype: bool
-    :raises CustomHTTException: If the user is not authorized to access the resource.
+    :raises CustomHTTPException: If the user is not authorized to access the resource.
     """
 
     def __init__(self, key: str):
@@ -234,15 +234,15 @@ def __init__(self, key: str):
     async def __call__(self, request: Request) -> str:
         authorization = request.headers.get("Authorization")
         if not authorization or not authorization.startswith("Bearer "):
-            raise CustomHTTException(
+            raise CustomHTTPException(
                 code_error=AuthErrorCode.AUTH_INVALID_TOKEN,
                 message_error="Invalid or missing token.",
                 status_code=status.HTTP_401_UNAUTHORIZED,
             )
         token = authorization.split("Bearer ")[1]
 
         if not (value := request.path_params.get(self.key) or request.query_params.get(self.key)):
-            raise CustomHTTException(
+            raise CustomHTTPException(
                 code_error=UserErrorCode.USER_NOT_FOUND,
                 message_error=f"Resource '{value}' not found.",
                 status_code=status.HTTP_400_BAD_REQUEST,
@@ -257,7 +257,7 @@ async def __call__(self, request: Request) -> str:
         ):
             return value
         else:
-            raise CustomHTTException(
+            raise CustomHTTPException(
                 code_error=UserErrorCode.USER_UNAUTHORIZED_PERFORM_ACTION,
                 message_error="You are not authorized to perform this action.",
                 status_code=status.HTTP_403_FORBIDDEN,

src/models/mixins.py

@@ -1,7 +1,8 @@
 from datetime import datetime, UTC
 from typing import Optional
+from pydantic import Field
 
 
 class DatetimeTimestamp:
-    created_at: Optional[datetime] = datetime.now(tz=UTC)
-    updated_at: Optional[datetime] = datetime.now(tz=UTC)
+    created_at: Optional[datetime] = Field(default=datetime.now(tz=UTC), description="Datetime created")
+    updated_at: Optional[datetime] = Field(default=datetime.now(tz=UTC), description="Datetime updated")

src/models/params.py

@@ -5,15 +5,16 @@
 from slugify import slugify
 from starlette import status
 
-from src.common.helpers.exceptions import CustomHTTException
+from pydantic import Field
+from src.common.helpers.exception import CustomHTTPException
 from src.config import settings
 from src.schemas import ParamsModel
 from src.shared.error_codes import ParamErrorCode
 from .mixins import DatetimeTimestamp
 
 
 class Params(ParamsModel, DatetimeTimestamp, Document):
-    slug: Optional[Indexed(str)] = None
+    slug: Optional[Indexed(str)] = Field(..., description="Slug of the parameter")
 
     class Settings:
         name = settings.PARAM_MODEL_NAME
@@ -34,7 +35,7 @@ class Settings:
     async def generate_unique_slug(self, **kwargs):
         new_slug_value = slugify(self.name)
         if await Params.find({"slug": new_slug_value, "type": self.type}).exists() is True:
-            raise CustomHTTException(
+            raise CustomHTTPException(
                 code_error=ParamErrorCode.PARAM_ALREADY_EXIST,
                 message_error=f"This parameter '{self.name}' already exists.",
                 status_code=status.HTTP_400_BAD_REQUEST,

src/models/roles.py

@@ -1,21 +1,21 @@
 from typing import Dict, List, Optional
 
 import pymongo
-from beanie import Document, Indexed, Insert, before_event
+from beanie import before_event, Document, Indexed, Insert
 from slugify import slugify
 from starlette import status
+from pydantic import Field
 
-from src.common.helpers.exceptions import CustomHTTException
+from src.common.helpers.exception import CustomHTTPException
 from src.config import settings
 from src.schemas import RoleModel
 from src.shared.error_codes import RoleErrorCode
-
 from .mixins import DatetimeTimestamp
 
 
 class Role(RoleModel, DatetimeTimestamp, Document):
-    permissions: List[Dict] = []
-    slug: Optional[Indexed(str)] = None
+    permissions: List[Dict] = Field(default_factory=list, description="Role permissions")
+    slug: Optional[Indexed(str)] = Field(None, description="Role slug")
 
     class Settings:
         name = settings.ROLE_MODEL_NAME
@@ -34,7 +34,7 @@ class Settings:
     async def generate_unique_slug(self, **kwargs):
         new_slug_value = slugify(self.name)
         if await Role.find({"slug": new_slug_value}).exists() is True:
-            raise CustomHTTException(
+            raise CustomHTTPException(
                 code_error=RoleErrorCode.ROLE_ALREADY_EXIST,
                 message_error=f"This role '{self.name}' already exists.",
                 status_code=status.HTTP_400_BAD_REQUEST,

src/models/users.py

@@ -2,15 +2,16 @@
 
 import pymongo
 from beanie import Document
+from pydantic import Field
 
 from src.config import settings
 from src.schemas import CreateUser
 from .mixins import DatetimeTimestamp
 
 
 class User(CreateUser, DatetimeTimestamp, Document):
-    is_active: Optional[bool] = False
-    is_primary: Optional[bool] = False
+    is_active: Optional[bool] = Field(False, description="User is active")
+    is_primary: Optional[bool] = Field(False, description="User is primary")
 
     class Settings:
         name = settings.USER_MODEL_NAME

src/routers/params.py

@@ -1,18 +1,17 @@
 from typing import Optional
 
 from beanie import PydanticObjectId
-from fastapi import APIRouter, Depends, Body, Query, status
+from fastapi import APIRouter, Body, Depends, Query, status
 from fastapi_pagination.ext.beanie import paginate
 from pymongo import ASCENDING, DESCENDING
 
+from src.common.helpers.pagination import customize_page
 from src.config import settings
+from src.middleware import AuthorizedHTTPBearer, CheckPermissionsHandler
 from src.models import Params
-from src.schemas.mixins import get_filter_params
-from src.schemas import ParamsModel, FilterParams
+from src.schemas import FilterParams, ParamsModel
 from src.services import params
-from src.shared.utils import customize_page, SortEnum
-from src.middleware import AuthorizedHTTPBearer, CheckPermissionsHandler
-
+from src.shared.utils import SortEnum
 
 param_router = APIRouter(prefix="/parameters", tags=["PARAMETERS"])
 
@@ -46,7 +45,7 @@ async def create(payload: ParamsModel = Body(...)):
     status_code=status.HTTP_200_OK,
 )
 async def all(
-    filter: FilterParams = Depends(get_filter_params),
+    filter: FilterParams = Depends(FilterParams),
     sort: Optional[SortEnum] = Query(default=SortEnum.DESC, alias="sort", description="Sort by 'asc' or 'desc"),
 ):
     search = {}