Integrate translation within the handler chain

viren-nadkarni · viren-nadkarni · commit fd07dd09eca4 · 2025-09-24T18:43:42.000+05:30
diff --git a/localstack-core/localstack/aws/handlers/logging.py b/localstack-core/localstack/aws/handlers/logging.py
@@ -1,5 +1,7 @@
 """Handlers for logging."""
 
+import contextlib
+import importlib
 import logging
 from functools import cached_property
 
@@ -21,14 +23,20 @@ class ExceptionLogger(ExceptionHandler):
     def __init__(self, logger=None):
         self.logger = logger or LOG
 
+        self._moto_service_exception = object
+        with contextlib.suppress(ModuleNotFoundError, AttributeError):
+            self._moto_service_exception = importlib.import_module(
+                "moto.core.exceptions"
+            ).ServiceException
+
     def __call__(
         self,
         chain: HandlerChain,
         exception: Exception,
         context: RequestContext,
         response: Response,
     ):
-        if isinstance(exception, ServiceException):
+        if isinstance(exception, (ServiceException, self._moto_service_exception)):
             # We do not want to log an error/stacktrace if the handler is working as expected, but chooses to throw
             # a service exception
             return
diff --git a/localstack-core/localstack/aws/handlers/service.py b/localstack-core/localstack/aws/handlers/service.py
@@ -1,5 +1,6 @@
 """A set of common handlers to parse and route AWS service requests."""
 
+import importlib
 import logging
 import traceback
 from collections import defaultdict
@@ -156,6 +157,14 @@ class ServiceExceptionSerializer(ExceptionHandler):
     def __init__(self):
         self.handle_internal_failures = True
 
+        self._moto_service_exception = object
+        try:
+            self._moto_service_exception = importlib.import_module(
+                "moto.core.exceptions"
+            ).ServiceException
+        except (ModuleNotFoundError, AttributeError) as exc:
+            LOG.debug("Unable to set up Moto ServiceException translation: %s", exc)
+
     def __call__(
         self,
         chain: HandlerChain,
@@ -183,6 +192,15 @@ def create_exception_response(self, exception: Exception, context: RequestContex
             LOG.info(message)
             context.service_exception = error
 
+        elif isinstance(exception, self._moto_service_exception):
+            # Parse Moto ServiceException native  if Moto is available.
+            # This allows handler chain to gracefully handles Moto errors when provider handlers invoke Moto methods directly.
+            # Moto may not be available in stripped-down versions of LocalStack, like LocalStack S3 image.
+            error = CommonServiceException(
+                code=exception.code,
+                message=exception.message,
+            )
+
         elif not isinstance(exception, ServiceException):
             if not self.handle_internal_failures:
                 return
