Feat/code refactor

app/config/__init__.py

@@ -1,7 +1,8 @@
+from __future__ import annotations
+
 from .redis_config import get_redis_pool
 
 
 __all__ = [
-    "engine",
     "get_redis_pool",
 ]

app/daos/home.py

@@ -1,14 +1,18 @@
+from __future__ import annotations
+
 import asyncio
 import random
 
+from app.exceptions import ExternalServiceException
+
 
 async def external_service_call():
     # Simulate network delay
-    delay = random.uniform(0.1, 1.0)  # Random delay between 0.1 to 1.0 seconds
+    delay = random.uniform(0.1, 1.0)  # Random delay between 0.1 to 1.0 seconds  #NOSONAR
     await asyncio.sleep(delay)
 
     # Simulate occasional failures
-    if random.random() < 0.2:  # 20% chance of failure
-        raise Exception("External service failed")
+    if random.random() < 0.2:  # 20% chance of failure #NOSONAR
+        raise ExternalServiceException("External Service Failed")
 
     return "Success from external service"

app/daos/users.py

@@ -1,27 +1,37 @@
+from __future__ import annotations
+
 import json
 
 from fastapi import HTTPException
+from fastapi_pagination.ext.sqlalchemy import paginate
+from sqlalchemy import select
 from sqlalchemy.orm import Session
+from werkzeug.security import check_password_hash
+
 from app.constants import jwt_utils
 from app.constants.messages.users import user_messages as messages
+from app.exceptions import EmailAlreadyExistException
+from app.exceptions import InvalidCredentialsException
+from app.exceptions import MobileAlreadyExistException
+from app.exceptions import NoUserFoundException
 from app.models import User
-from app.schemas.users.users_request import CreateUser, Login
-from werkzeug.security import check_password_hash
-from fastapi_pagination.ext.sqlalchemy import paginate
-from sqlalchemy import select
-from app.utils.user_utils import check_existing_field, responseFormatter
-from app.wrappers.cache_wrappers import create_cache, retrieve_cache
+from app.schemas.users.users_request import CreateUser
+from app.schemas.users.users_request import Login
+from app.utils.user_utils import check_existing_field
+from app.utils.user_utils import response_formatter
+from app.wrappers.cache_wrappers import create_cache
+from app.wrappers.cache_wrappers import retrieve_cache
 
 
-async def get_user(user_id: int, dbSession: Session):
+async def get_user(user_id: int, db_session: Session):
     try:
         cache_key = f"user_{user_id}"
-        cached_user, expire = await retrieve_cache(cache_key)
+        cached_user, _ = await retrieve_cache(cache_key)
         if cached_user:
             return json.loads(cached_user)
         # Check if the user already exists in the database
         user = (
-            dbSession.query(User)
+            db_session.query(User)
             .where(User.id == user_id)
             .with_entities(
                 User.id,
@@ -35,7 +45,7 @@ async def get_user(user_id: int, dbSession: Session):
             .first()
         )
         if not user:
-            raise Exception(messages["NO_USER_FOUND_FOR_ID"])
+            raise NoUserFoundException(messages['NO_USER_FOUND_FOR_ID'])
 
         await create_cache(json.dumps(user._asdict(), default=str), cache_key, 60)
         return user
@@ -45,12 +55,12 @@ async def get_user(user_id: int, dbSession: Session):
         raise HTTPException(status_code=400, detail=f"{str(e)}")
 
 
-def list_users(dbSession: Session):
+def list_users(db_session: Session):
     try:
         query = select(User.id, User.name, User.email, User.mobile).order_by(User.created_at)
 
         # Pass the Select object to the paginate function
-        users = paginate(dbSession, query=query)
+        users = paginate(db_session, query=query)
 
         return users
 
@@ -60,54 +70,59 @@ def list_users(dbSession: Session):
         raise HTTPException(status_code=400, detail=f"{str(e)}")
 
 
-def create_user(data: CreateUser, dbSession: Session):
+def create_user(data: CreateUser, db_session: Session):
     try:
         user_data = data.dict()
         # Check if the email already exists in the db
-        email_exists = check_existing_field(dbSession=dbSession, model=User, field="email", value=user_data["email"])
+        email_exists = check_existing_field(db_session=db_session, model=User, field="email", value=user_data["email"])
         if email_exists:
-            raise Exception(messages["EMAIL_ALREADY_EXIST"])
+            raise EmailAlreadyExistException(messages['EMAIL_ALREADY_EXIST'])
 
         # Check if the mobile already exists in the db
-        mobile_exists = check_existing_field(dbSession=dbSession, model=User, field="mobile", value=user_data["mobile"])
+        mobile_exists = check_existing_field(
+            db_session=db_session,
+            model=User,
+            field="mobile",
+            value=user_data["mobile"],
+        )
         if mobile_exists:
-            raise Exception(messages["MOBILE_ALREADY_EXIST"])
+            raise MobileAlreadyExistException(messages['MOBILE_ALREADY_EXIST'])
 
         user = User(**user_data)
 
-        dbSession.add(user)
-        dbSession.commit()
-        dbSession.refresh(user)
+        db_session.add(user)
+        db_session.commit()
+        db_session.refresh(user)
 
-        return responseFormatter(messages["CREATED_SUCCESSFULLY"])
+        return response_formatter(messages["CREATED_SUCCESSFULLY"])
 
     except Exception as e:
         # Return a user-friendly error message to the client
         raise HTTPException(status_code=400, detail=f"{str(e)}")
 
 
-def login(data: Login, dbSession: Session):
+def login(data: Login, db_session: Session):
     try:
         user_data = data.dict()
 
         # Check if the course already exists in the db
         user_details = (
-            dbSession.query(User)
+            db_session.query(User)
             .where(
                 User.email == user_data["email"],
             )
             .first()
         )
 
         if not user_details:
-            raise Exception(messages["INVALID_CREDENTIALS"])
+            raise InvalidCredentialsException(messages['INVALID_CREDENTIALS'])
 
         if not check_password_hash(user_details.password, user_data["password"]):
-            raise Exception(messages["INVALID_CREDENTIALS"])
+            raise InvalidCredentialsException(messages['INVALID_CREDENTIALS'])
 
         del user_details.password
         token = jwt_utils.create_access_token({"sub": user_details.email, "id": user_details.id})
-        return responseFormatter(messages["LOGIN_SUCCESSFULLY"], {"token": token})
+        return response_formatter(messages["LOGIN_SUCCESSFULLY"], {"token": token})
 
     except Exception as e:
         print(e)

app/exceptions.py

@@ -0,0 +1,24 @@
+class ExternalServiceException(Exception):
+    pass
+
+
+class NoUserFoundException(Exception):
+    pass
+class EmailAlreadyExistException(Exception):
+    pass
+
+class MobileAlreadyExistException(Exception):
+    pass
+
+class InvalidCredentialsException(Exception):
+    pass
+
+class CentryTestException(Exception):
+    pass
+
+class DatabaseConnectionException(Exception):
+    pass
+
+
+class RedisUrlNotFoundException(Exception):
+    pass

app/middlewares/cache_middleware.py

@@ -15,9 +15,9 @@
 
 class CacheMiddleware(BaseHTTPMiddleware):
     def __init__(
-        self,
-        app,
-        cached_endpoints: list[str],
+            self,
+            app,
+            cached_endpoints: list[str],
     ):
         super().__init__(app)
         self.cached_endpoints = cached_endpoints
@@ -28,6 +28,10 @@ def matches_any_path(self, path_url):
                 return True
         return False
 
+    async def handle_max_age(self, max_age, response_body, key):
+        if max_age:
+            await create_cache(response_body[0].decode(), key, max_age)
+
     async def dispatch(self, request: Request, call_next) -> Response:
         path_url = request.url.path
         request_type = request.method
@@ -58,8 +62,7 @@ async def dispatch(self, request: Request, call_next) -> Response:
                 max_age_match = re.search(r"max-age=(\d+)", cache_control)
                 if max_age_match:
                     max_age = int(max_age_match.group(1))
-                    if max_age:
-                        await create_cache(response_body[0].decode(), key, max_age)
+                    await self.handle_max_age(max_age, response_body, key)
             elif matches:
                 await create_cache(response_body[0].decode(), key, max_age)
         return response

app/middlewares/rate_limiter_middleware.py

@@ -26,7 +26,7 @@ async def dispatch(self, request: Request, call_next):
             pipe.expire(client_ip, TIME_WINDOW)
             await pipe.execute()
         finally:
-            pass
+            print("Finally Block in Rate Limit exceeded")
 
         response = await call_next(request)
         return response

app/middlewares/request_id_injection.py

@@ -1,17 +1,19 @@
+from __future__ import annotations
+
 import contextvars
 import uuid
 
+from fastapi.responses import JSONResponse
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.requests import Request
-from fastapi.responses import JSONResponse
 
 request_id_contextvar = contextvars.ContextVar("request_id", default=None)
 
 
 class RequestIdInjection(BaseHTTPMiddleware):
     def dispatch(self, request: Request, call_next):
         request_id = str(uuid.uuid4())
-        request_id_contextvar.set(request_id)
+        request_id_contextvar.set(request_id)  # noqa
         try:
             return call_next(request)
 

app/models/__init__.py

@@ -1,12 +1,4 @@
-# import glob
-# from os.path import basename
-# from os.path import dirname
-# from os.path import isfile
-# from os.path import join
-
-# modules = glob.glob(join(dirname(__file__), "*.py"))
-
-# __all__ = [basename(f)[:-3] for f in modules if isfile(f) and not f.endswith("__init__.py")]
+from __future__ import annotations
 
 from .users import User
 

app/models/users.py

@@ -1,17 +1,20 @@
-from sqlalchemy import Column, event
+from __future__ import annotations
+
+from sqlalchemy import Column
+from sqlalchemy import event
+from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.sql import func
 from sqlalchemy.sql.sqltypes import DateTime
 from sqlalchemy.sql.sqltypes import Integer
 from sqlalchemy.sql.sqltypes import String
-from sqlalchemy.ext.declarative import declarative_base
 from werkzeug.security import generate_password_hash
 
 from app.sessions.db import engine
 
 Base = declarative_base()
 
 
-class User(Base):
+class User(Base):  # noqa
     __tablename__ = "user"
 
     id = Column(Integer, primary_key=True, index=True)

app/routes/cache_router/cache_samples.py

@@ -11,5 +11,5 @@
 @cache_sample_router.get("/get-cache", tags=["Cache-Sample"])
 def get_cache():
     print("Request ID:", request_id_contextvar.get())
-    response = random.randint(100, 1000)
+    response = random.randint(100, 1000) #NOSONAR
     return {"random value is": response}

app/routes/home/home.py

@@ -1,10 +1,15 @@
-from dependencies import circuit_breaker
-from fastapi import APIRouter, HTTPException
+from __future__ import annotations
+
+from fastapi import APIRouter
+from fastapi import HTTPException
 from fastapi.responses import JSONResponse
+from pybreaker import CircuitBreakerError
+
 from app.config.base import settings
-from app.middlewares.request_id_injection import request_id_contextvar
 from app.daos.home import external_service_call
-from pybreaker import CircuitBreakerError
+from app.exceptions import CentryTestException
+from app.middlewares.request_id_injection import request_id_contextvar
+from dependencies import circuit_breaker
 
 home_router = APIRouter()
 
@@ -32,7 +37,7 @@ async def external_service_endpoint():
 def sentry_endpoint():
     if not settings.SENTRY_DSN:
         raise HTTPException(status_code=503, detail="Sentry DSN not found")
-    raise Exception("Testing Sentry")
+    raise CentryTestException("Centry Test")
 
 
 @home_router.get("/{path:path}", include_in_schema=False)