Merge pull request #58 from Mohammadreza-kh94/feature/grpc-exception-interceptor

Feature/grpc exception interceptor

Author: HosseinNejatiJavaremi

archipy/configs/base_config.py

@@ -31,6 +31,7 @@
     StarRocksSQLAlchemyConfig,
 )
 from archipy.configs.environment_type import EnvironmentType
+from archipy.models.types import LanguageType
 
 """
 
@@ -171,6 +172,7 @@ def settings_customise_sources(
     STARROCKS_SQLALCHEMY: StarRocksSQLAlchemyConfig = StarRocksSQLAlchemyConfig()
     POSTGRES_SQLALCHEMY: PostgresSQLAlchemyConfig = PostgresSQLAlchemyConfig()
     SQLITE_SQLALCHEMY: SQLiteSQLAlchemyConfig = SQLiteSQLAlchemyConfig()
+    LANGUAGE: LanguageType = LanguageType.FA
 
     def customize(self) -> None:
         """Customize configuration after loading.

archipy/helpers/decorators/retry.py

@@ -4,7 +4,6 @@
 from typing import Any, TypeVar, cast
 
 from archipy.models.errors import ResourceExhaustedError
-from archipy.models.types.language_type import LanguageType
 
 # Define a type variable for the return type of the decorated function
 F = TypeVar("F", bound=Callable[..., Any])
@@ -16,7 +15,6 @@ def retry_decorator(
     retry_on: tuple[type[Exception], ...] | None = None,
     ignore: tuple[type[Exception], ...] | None = None,
     resource_type: str | None = None,
-    lang: LanguageType = LanguageType.FA,
 ) -> Callable[[F], F]:
     """A decorator that retries a function when it raises an exception.
 
@@ -75,7 +73,7 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
                         time.sleep(delay)
                     continue
                 return result
-            raise ResourceExhaustedError(resource_type=resource_type, lang=lang)
+            raise ResourceExhaustedError(resource_type=resource_type)
 
         return cast(F, wrapper)
 

archipy/helpers/interceptors/grpc/exception/__init__.py

@@ -0,0 +1,8 @@
+"""Exception handling interceptors for gRPC services."""
+
+from .server_interceptor import AsyncGrpcServerExceptionInterceptor, GrpcServerExceptionInterceptor
+
+__all__ = [
+    "GrpcServerExceptionInterceptor",
+    "AsyncGrpcServerExceptionInterceptor",
+]

archipy/helpers/interceptors/grpc/exception/server_interceptor.py

@@ -0,0 +1,209 @@
+from collections.abc import Callable
+
+import grpc
+from pydantic import ValidationError
+
+from archipy.helpers.interceptors.grpc.base.server_interceptor import (
+    BaseAsyncGrpcServerInterceptor,
+    BaseGrpcServerInterceptor,
+    MethodName,
+)
+from archipy.helpers.utils.base_utils import BaseUtils
+from archipy.models.errors import InternalError, InvalidArgumentError
+from archipy.models.errors.base_error import BaseError
+
+
+class GrpcServerExceptionInterceptor(BaseGrpcServerInterceptor):
+    """A sync gRPC server interceptor for centralized exception handling.
+
+    This interceptor catches all exceptions thrown by gRPC service methods and
+    converts them to appropriate gRPC errors, eliminating the need for repetitive
+    try-catch blocks in each service method.
+    """
+
+    def intercept(
+        self, method: Callable, request: object, context: grpc.ServicerContext, method_name_model: MethodName
+    ) -> object:
+        """Intercepts a sync gRPC server call and handles exceptions.
+
+        Args:
+            method: The sync gRPC method being intercepted.
+            request: The request object passed to the method.
+            context: The context of the sync gRPC call.
+            method_name_model: The parsed method name containing package, service, and method components.
+
+        Returns:
+            object: The result of the intercepted gRPC method.
+
+        Note:
+            This method will not return anything if an exception is handled,
+            as the exception handling will abort the gRPC context.
+        """
+        try:
+            # Execute the gRPC method
+            result = method(request, context)
+
+        except ValidationError as validation_error:
+            BaseUtils.capture_exception(validation_error)
+            self._handle_validation_error(validation_error, context)
+
+        except BaseError as base_error:
+            BaseUtils.capture_exception(base_error)
+            base_error.abort_grpc_sync(context)
+
+        except Exception as unexpected_error:
+            BaseUtils.capture_exception(unexpected_error)
+            self._handle_unexpected_error(unexpected_error, context, method_name_model)
+        else:
+            return result
+
+    @staticmethod
+    def _handle_validation_error(validation_error: ValidationError, context: grpc.ServicerContext) -> None:
+        """Handle Pydantic validation errors.
+
+        Args:
+            validation_error: The validation error to handle.
+            context: The gRPC context to abort.
+        """
+        # Format validation errors for better debugging
+        validation_details = BaseUtils.format_validation_errors(validation_error, include_type=True)
+
+        InvalidArgumentError(
+            argument_name="request_validation",
+            additional_data={"validation_errors": validation_details, "error_count": len(validation_error.errors())},
+        ).abort_grpc_sync(context)
+
+    @staticmethod
+    def _handle_unexpected_error(
+        error: Exception, context: grpc.ServicerContext, method_name_model: MethodName
+    ) -> None:
+        """Handle unexpected errors by converting them to internal errors.
+
+        Args:
+            error: The unexpected error to handle.
+            context: The gRPC context to abort.
+            method_name_model: The method name information for better error tracking.
+        """
+        # Capture the exception for monitoring
+        InternalError(
+            additional_data={
+                "original_error": str(error),
+                "error_type": type(error).__name__,
+                "service": method_name_model.service,
+                "method": method_name_model.method,
+                "package": method_name_model.package,
+            }
+        ).abort_grpc_sync(context)
+
+    @staticmethod
+    def _format_validation_errors(validation_error: ValidationError) -> list[dict[str, str]]:
+        """Format Pydantic validation errors into a structured format.
+
+        Args:
+            validation_error: The validation error to format.
+
+        Returns:
+            A list of formatted validation error details.
+
+        Note:
+            This method is deprecated. Use BaseUtils.format_validation_errors instead.
+        """
+        return BaseUtils.format_validation_errors(validation_error, include_type=True)
+
+
+class AsyncGrpcServerExceptionInterceptor(BaseAsyncGrpcServerInterceptor):
+    """An async gRPC server interceptor for centralized exception handling.
+
+    This interceptor catches all exceptions thrown by gRPC service methods and
+    converts them to appropriate gRPC errors, eliminating the need for repetitive
+    try-catch blocks in each service method.
+    """
+
+    async def intercept(
+        self, method: Callable, request: object, context: grpc.aio.ServicerContext, method_name_model: MethodName
+    ) -> object:
+        """Intercepts an async gRPC server call and handles exceptions.
+
+        Args:
+            method: The async gRPC method being intercepted.
+            request: The request object passed to the method.
+            context: The context of the async gRPC call.
+            method_name_model: The parsed method name containing package, service, and method components.
+
+        Returns:
+            object: The result of the intercepted gRPC method.
+
+        Note:
+            This method will not return anything if an exception is handled,
+            as the exception handling will abort the gRPC context.
+        """
+        try:
+            # Execute the gRPC method
+            result = await method(request, context)
+
+        except ValidationError as validation_error:
+            BaseUtils.capture_exception(validation_error)
+            await self._handle_validation_error(validation_error, context)
+
+        except BaseError as base_error:
+            BaseUtils.capture_exception(base_error)
+            await base_error.abort_grpc_async(context)
+
+        except Exception as unexpected_error:
+            BaseUtils.capture_exception(unexpected_error)
+            await self._handle_unexpected_error(unexpected_error, context, method_name_model)
+        else:
+            return result
+
+    @staticmethod
+    async def _handle_validation_error(validation_error: ValidationError, context: grpc.aio.ServicerContext) -> None:
+        """Handle Pydantic validation errors.
+
+        Args:
+            validation_error: The validation error to handle.
+            context: The gRPC context to abort.
+        """
+        # Format validation errors for better debugging
+        validation_details = BaseUtils.format_validation_errors(validation_error, include_type=True)
+
+        await InvalidArgumentError(
+            argument_name="request_validation",
+            additional_data={"validation_errors": validation_details, "error_count": len(validation_error.errors())},
+        ).abort_grpc_async(context)
+
+    @staticmethod
+    async def _handle_unexpected_error(
+        error: Exception, context: grpc.aio.ServicerContext, method_name_model: MethodName
+    ) -> None:
+        """Handle unexpected errors by converting them to internal errors.
+
+        Args:
+            error: The unexpected error to handle.
+            context: The gRPC context to abort.
+            method_name_model: The method name information for better error tracking.
+        """
+        # Capture the exception for monitoring
+        await InternalError(
+            additional_data={
+                "original_error": str(error),
+                "error_type": type(error).__name__,
+                "service": method_name_model.service,
+                "method": method_name_model.method,
+                "package": method_name_model.package,
+            }
+        ).abort_grpc_async(context)
+
+    @staticmethod
+    def _format_validation_errors(validation_error: ValidationError) -> list[dict[str, str]]:
+        """Format Pydantic validation errors into a structured format.
+
+        Args:
+            validation_error: The validation error to format.
+
+        Returns:
+            A list of formatted validation error details.
+
+        Note:
+            This method is deprecated. Use BaseUtils.format_validation_errors instead.
+        """
+        return BaseUtils.format_validation_errors(validation_error, include_type=True)

archipy/helpers/utils/app_utils.py

@@ -39,7 +39,9 @@ def create_error_response(exception: BaseError) -> JSONResponse:
         BaseUtils.capture_exception(exception)
         # Default to internal server error if status code is not set
         status_code = (
-            exception.http_status_code_value if exception.http_status_code_value else HTTPStatus.INTERNAL_SERVER_ERROR.value
+            exception.http_status_code_value
+            if exception.http_status_code_value
+            else HTTPStatus.INTERNAL_SERVER_ERROR.value
         )
         return JSONResponse(status_code=status_code, content=exception.to_dict())
 
@@ -85,16 +87,8 @@ async def validation_exception_handler(
         Returns:
             JSONResponse: A JSON response containing the validation error details.
         """
-        # Using list comprehension instead of append for better performance
         BaseUtils.capture_exception(exception)
-        errors: list[dict[str, str]] = [
-            {
-                "field": ".".join(str(x) for x in error["loc"]),
-                "message": error["msg"],
-                "value": str(error.get("input", "")),
-            }
-            for error in exception.errors()
-        ]
+        errors = BaseUtils.format_validation_errors(exception)
         return JSONResponse(
             status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
             content={"error": "VALIDATION_ERROR", "detail": errors},

archipy/helpers/utils/base_utils.py

@@ -1,5 +1,7 @@
 import re
 
+from pydantic import ValidationError
+
 from archipy.helpers.utils.datetime_utils import DatetimeUtils
 from archipy.helpers.utils.error_utils import ErrorUtils
 from archipy.helpers.utils.file_utils import FileUtils
@@ -20,6 +22,32 @@ class BaseUtils(ErrorUtils, DatetimeUtils, PasswordUtils, JWTUtils, TOTPUtils, F
     This class inherits from various utility classes to provide a centralized place for common utility methods.
     """
 
+    @staticmethod
+    def format_validation_errors(
+        validation_error: ValidationError, *, include_type: bool = False
+    ) -> list[dict[str, str]]:
+        """Formats Pydantic validation errors into a structured format.
+
+        Args:
+            validation_error (ValidationError): The validation error to format.
+            include_type (bool): Whether to include the error type in the output. Defaults to False.
+
+        Returns:
+            list[dict[str, str]]: A list of formatted validation error details.
+        """
+        formatted_errors = []
+        for error in validation_error.errors():
+            error_dict = {
+                "field": ".".join(str(x) for x in error["loc"]),
+                "message": error["msg"],
+                "value": str(error.get("input", "")),
+            }
+            if include_type:
+                error_dict["type"] = error["type"]
+            formatted_errors.append(error_dict)
+
+        return formatted_errors
+
     @staticmethod
     def sanitize_iranian_landline_or_phone_number(landline_or_phone_number: str) -> str:
         """Sanitizes an Iranian landline or mobile phone number by removing non-numeric characters and standardizing the format.
@@ -85,6 +113,7 @@ def validate_iranian_landline_number(cls, landline_number: str) -> None:
     @classmethod
     def validate_iranian_national_code_pattern(cls, national_code: str) -> None:
         """Validates an Iranian National ID number using the official algorithm.
+
         To see how the algorithm works, see http://www.aliarash.com/article/codemeli/codemeli.htm
 
         The algorithm works by:

archipy/helpers/utils/password_utils.py

@@ -66,7 +66,6 @@ def verify_password(password: str, stored_password: str, auth_config: AuthConfig
     def validate_password(
         password: str,
         auth_config: AuthConfig | None = None,
-        lang: LanguageType = LanguageType.FA,
     ) -> None:
         """Validates a password against the password policy.
 
@@ -97,7 +96,7 @@ def validate_password(
             errors.append(f"Password must contain at least one special character: {configs.SPECIAL_CHARACTERS}")
 
         if errors:
-            raise InvalidPasswordError(requirements=errors, lang=lang)
+            raise InvalidPasswordError(requirements=errors)
 
     @staticmethod
     def generate_password(auth_config: AuthConfig | None = None) -> str:
@@ -146,7 +145,7 @@ def validate_password_history(
         new_password: str,
         password_history: list[str],
         auth_config: AuthConfig | None = None,
-        lang: LanguageType = LanguageType.FA,
+        lang: LanguageType | None = None,
     ) -> None:
         """Validates a new password against the password history.