fix(utils): resolve all MyPy and Ruff errors in helpers/utils

Author: HosseinNejatiJavaremi

archipy/helpers/utils/app_utils.py

@@ -166,7 +166,7 @@ def setup_elastic_apm(app: FastAPI, config: BaseConfig) -> None:
             from elasticapm.contrib.starlette import ElasticAPM, make_apm_client
 
             apm_client = make_apm_client(config.ELASTIC_APM.model_dump())
-            app.add_middleware(ElasticAPM, client=apm_client)
+            app.add_middleware(ElasticAPM, client=apm_client)  # type: ignore[arg-type]
         except Exception:
             logging.exception("Failed to initialize Elastic APM")
 

archipy/helpers/utils/keycloak_utils.py

@@ -5,7 +5,6 @@
 from typing import Any
 
 try:
-    import grpc
     from grpc import ServicerContext
     from grpc.aio import ServicerContext as AsyncServicerContext
 
@@ -116,18 +115,21 @@ def dependency(
             request: Request,
             token: HTTPAuthorizationCredentials = Security(security),
             keycloak: KeycloakAdapter = Depends(cls._get_keycloak_adapter),
-        ) -> dict:
+        ) -> dict[str, Any]:
             if token is None:
                 raise UnauthenticatedError(lang=lang)
             token_str = token.credentials  # Extract the token string
             # Validate token
             if not keycloak.validate_token(token_str):
                 token_info = keycloak.introspect_token(token_str)
-                if not token_info.get("active", False):
+                if not token_info or not token_info.get("active", False):
                     raise TokenExpiredError(lang=lang)
 
             # Get user info from token
             user_info = keycloak.get_userinfo(token_str)
+            if not user_info:
+                raise UnauthenticatedError(lang=lang)
+
             token_info = keycloak.get_token_info(token_str)
 
             # Resource-based authorization if resource type is provided
@@ -219,7 +221,7 @@ async def dependency(
             request: Request,
             token: HTTPAuthorizationCredentials = Security(security),
             keycloak: AsyncKeycloakAdapter = Depends(cls._get_async_keycloak_adapter),
-        ) -> dict:
+        ) -> dict[str, Any]:
             if token is None:
                 raise UnauthenticatedError(lang=lang)
             token_str = token.credentials  # Extract the token string
@@ -228,11 +230,14 @@ async def dependency(
             if not await keycloak.validate_token(token_str):
                 # Handle token validation error
                 token_info = await keycloak.introspect_token(token_str)
-                if not token_info.get("active", False):
+                if not token_info or not token_info.get("active", False):
                     raise TokenExpiredError(lang=lang)
 
             # Get user info from token
             user_info = await keycloak.get_userinfo(token_str)
+            if not user_info:
+                raise UnauthenticatedError(lang=lang)
+
             token_info = await keycloak.get_token_info(token_str)
 
             # Resource-based authorization if resource type is provided
@@ -287,6 +292,8 @@ async def dependency(
             # Add user info to request state
             request.state.user_info = user_info
             request.state.token_info = token_info
+            if not user_info:
+                raise UnauthenticatedError(lang=lang)
             return user_info
 
         return dependency
@@ -302,11 +309,11 @@ def _extract_token_from_metadata(context: ServicerContext) -> str | None:
             if key in metadata:
                 auth_value = metadata[key]
                 if auth_value.startswith("Bearer "):
-                    return auth_value[7:]
+                    return str(auth_value[7:])
                 elif auth_value.startswith("bearer "):
-                    return auth_value[7:]
+                    return str(auth_value[7:])
                 else:
-                    return auth_value
+                    return str(auth_value)
 
         return None
 
@@ -356,11 +363,13 @@ def wrapper(self: object, request: object, context: ServicerContext) -> object:
                     # 3. Validate token
                     if not keycloak.validate_token(token_str):
                         token_info = keycloak.introspect_token(token_str)
-                        if not token_info.get("active", False):
+                        if not token_info or not token_info.get("active", False):
                             raise TokenExpiredError(lang=lang)
 
                     # 4. Get user info from token
                     user_info = keycloak.get_userinfo(token_str)
+                    if not user_info:
+                        raise UnauthenticatedError(lang=lang)
 
                     # 5. Resource-based authorization if resource_attribute_name is provided
                     if resource_attribute_name:
@@ -494,11 +503,13 @@ async def wrapper(self: object, request: object, context: AsyncServicerContext)
                     # 3. Validate token
                     if not await keycloak.validate_token(token_str):
                         token_info = await keycloak.introspect_token(token_str)
-                        if not token_info.get("active", False):
+                        if not token_info or not token_info.get("active", False):
                             raise TokenExpiredError(lang=lang)
 
                     # 4. Get user info from token
                     user_info = await keycloak.get_userinfo(token_str)
+                    if not user_info:
+                        raise UnauthenticatedError(lang=lang)
 
                     # 5. Resource-based authorization if resource_attribute_name is provided
                     if resource_attribute_name:
@@ -573,10 +584,12 @@ async def wrapper(self: object, request: object, context: AsyncServicerContext)
                 except Exception as e:
                     if isinstance(e, BaseError):
                         await e.abort_grpc_async(context)
+                        return None  # abort_grpc_async will terminate, but satisfy type checker
                     await InternalError(
                         lang=lang,
                         additional_data={"original_error": str(e), "error_type": type(e).__name__},
                     ).abort_grpc_async(context)
+                    return None  # abort_grpc_async will terminate, but satisfy type checker
 
                 finally:
                     # Clean up auth context

archipy/helpers/utils/string_utils.py

@@ -1,5 +1,5 @@
 import re
-from re import compile
+from re import compile as re_compile
 
 from archipy.helpers.utils.string_utils_constants import StringUtilsConstants
 
@@ -169,7 +169,7 @@ def mask_urls(cls, text: str, mask: str | None = None) -> str:
             str: The text with URLs masked.
         """
         mask = mask or "MASK_URL"
-        return compile(r"https?://\S+|www\.\S+").sub(f" {mask} ", text)
+        return re_compile(r"https?://\S+|www\.\S+").sub(f" {mask} ", text)
 
     @classmethod
     def mask_emails(cls, text: str, mask: str | None = None) -> str:
@@ -183,7 +183,7 @@ def mask_emails(cls, text: str, mask: str | None = None) -> str:
             str: The text with email addresses masked.
         """
         mask = mask or "MASK_EMAIL"
-        return compile(r"\S+@\S+\.\S+").sub(f" {mask} ", text)
+        return re_compile(r"\S+@\S+\.\S+").sub(f" {mask} ", text)
 
     @classmethod
     def mask_phones(cls, text: str, mask: str | None = None) -> str:
@@ -197,7 +197,7 @@ def mask_phones(cls, text: str, mask: str | None = None) -> str:
             str: The text with phone numbers masked.
         """
         mask = mask or "MASK_PHONE"
-        return compile(r"(?:\+98|0)?(?:\d{3}\s*?\d{3}\s*?\d{4})").sub(f" {mask} ", text)
+        return re_compile(r"(?:\+98|0)?(?:\d{3}\s*?\d{3}\s*?\d{4})").sub(f" {mask} ", text)
 
     @classmethod
     def convert_english_number_to_persian(cls, text: str) -> str:
@@ -308,7 +308,7 @@ def replace_currencies_with_mask(cls, text: str, mask: str | None = None) -> str
             str: The text with currency symbols and amounts masked.
         """
         mask = mask or "MASK_CURRENCIES"
-        currency_pattern = compile(r"(\\|zł|£|\$|₡|₦|¥|₩|₪|₫|€|₱|₲|₴|₹|﷼)+")
+        currency_pattern = re_compile(r"(\\|zł|£|\$|₡|₦|¥|₩|₪|₫|€|₱|₲|₴|₹|﷼)+")
         return currency_pattern.sub(f" {mask} ", text)
 
     @classmethod

archipy/helpers/utils/string_utils_constants.py

@@ -1,11 +1,22 @@
-from re import compile
+import re
+from re import compile as re_compile
 
 
-def compile_patterns(patterns):
-    return [(compile(pattern), repl) for pattern, repl in patterns]
+def compile_patterns(patterns: list[tuple[str, str]]) -> list[tuple[re.Pattern[str], str]]:
+    """Compile regex patterns with their replacement strings.
+
+    Args:
+        patterns: List of tuples containing (pattern, replacement) pairs.
+
+    Returns:
+        List of tuples containing (compiled_pattern, replacement) pairs.
+    """
+    return [(re_compile(pattern), repl) for pattern, repl in patterns]
 
 
 class StringUtilsConstants:
+    """Constants for string utility operations including translation tables and regex patterns."""
+
     arabic_vowel_translate_table = str.maketrans(
         dict.fromkeys(
             "\u064e\u064f\u0650\u0652\u0651\u0653\u064b\u064c\u0621\u064d\u0670"  # Normal vowels (Fatha, Damma, Kasra, etc)