Speed up exception handling by re-using Environments

moto/cloudfront/exceptions.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from jinja2 import DictLoader, Environment
 from moto.core.exceptions import RESTError
 
 EXCEPTION_RESPONSE = """<?xml version="1.0"?>
@@ -15,10 +16,11 @@
 
 class CloudFrontException(RESTError):
     code = 400
+    extended_templates = {"cferror": EXCEPTION_RESPONSE}
+    env = Environment(loader=DictLoader(RESTError.templates | extended_templates))
 
     def __init__(self, error_type: str, message: str, **kwargs: Any):
         kwargs.setdefault("template", "cferror")
-        self.templates["cferror"] = EXCEPTION_RESPONSE
         super().__init__(error_type, message, **kwargs)
 
 

moto/core/exceptions.py

@@ -54,6 +54,7 @@ class RESTError(HTTPException):
         "wrapped_single_error": WRAPPED_SINGLE_ERROR_RESPONSE,
         "error": ERROR_RESPONSE,
     }
+    env = Environment(loader=DictLoader(templates))
 
     def __init__(
         self, error_type: str, message: str, template: str = "error", **kwargs: Any
@@ -63,8 +64,7 @@ def __init__(
         self.message = message
 
         if template in self.templates.keys():
-            env = Environment(loader=DictLoader(self.templates))
-            self.description: str = env.get_template(template).render(
+            self.description: str = self.env.get_template(template).render(
                 error_type=error_type,
                 message=message,
                 request_id_tag=self.request_id_tag_name,

moto/ec2/exceptions.py

@@ -1,5 +1,6 @@
 from typing import Any, Iterable, List, Optional, Union
 
+from jinja2 import DictLoader, Environment
 from moto.core.exceptions import RESTError
 
 # EC2 has a custom root-tag - <Response> vs <ErrorResponse>
@@ -22,10 +23,11 @@ class EC2ClientError(RESTError):
     code = 400
     # EC2 uses <RequestID> as tag name in the XML response
     request_id_tag_name = "RequestID"
+    extended_templates = {"custom_response": EC2_ERROR_RESPONSE}
+    env = Environment(loader=DictLoader(RESTError.templates | extended_templates))
 
     def __init__(self, *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "custom_response")
-        self.templates["custom_response"] = EC2_ERROR_RESPONSE
         super().__init__(*args, **kwargs)
 
 

moto/elasticache/exceptions.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from jinja2 import DictLoader, Environment
 from moto.core.exceptions import RESTError
 
 EXCEPTION_RESPONSE = """<?xml version="1.0"?>
@@ -16,6 +17,8 @@
 class ElastiCacheException(RESTError):
 
     code = 400
+    extended_templates = {"ecerror": EXCEPTION_RESPONSE}
+    env = Environment(loader=DictLoader(RESTError.templates | extended_templates))
 
     def __init__(self, code: str, message: str, **kwargs: Any):
         kwargs.setdefault("template", "ecerror")

moto/elasticbeanstalk/exceptions.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from jinja2 import DictLoader, Environment
 from moto.core.exceptions import RESTError
 
 EXCEPTION_RESPONSE = """<?xml version="1.0"?>
@@ -16,6 +17,8 @@
 class ElasticBeanstalkException(RESTError):
 
     code = 400
+    extended_templates = {"ecerror": EXCEPTION_RESPONSE}
+    env = Environment(loader=DictLoader(RESTError.templates | extended_templates))
 
     def __init__(self, code: str, message: str, **kwargs: Any):
         kwargs.setdefault("template", "ecerror")

moto/s3/exceptions.py

@@ -1,5 +1,6 @@
 from typing import TYPE_CHECKING, Any, Optional, Union
 
+from jinja2 import DictLoader, Environment
 from moto.core.exceptions import RESTError
 
 if TYPE_CHECKING:
@@ -41,6 +42,17 @@ class S3ClientError(RESTError):
     # S3 API uses <RequestID> as the XML tag in response messages
     request_id_tag_name = "RequestID"
 
+    extended_templates = {
+        "bucket_error": ERROR_WITH_BUCKET_NAME,
+        "key_error": ERROR_WITH_KEY_NAME,
+        "argument_error": ERROR_WITH_ARGUMENT,
+        "error_uploadid": ERROR_WITH_UPLOADID,
+        "condition_error": ERROR_WITH_CONDITION_NAME,
+        "range_error": ERROR_WITH_RANGE,
+        "storage_error": ERROR_WITH_STORAGE_CLASS,
+    }
+    env = Environment(loader=DictLoader(RESTError.templates | extended_templates))
+
     def __init__(self, *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "single_error")
         self.templates["bucket_error"] = ERROR_WITH_BUCKET_NAME

moto/s3control/exceptions.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from jinja2 import DictLoader, Environment
 from moto.core.exceptions import RESTError
 
 ERROR_WITH_ACCESS_POINT_NAME = """{% extends 'wrapped_single_error' %}
@@ -13,6 +14,12 @@
 
 
 class S3ControlError(RESTError):
+    extended_templates = {
+        "ap_not_found": ERROR_WITH_ACCESS_POINT_NAME,
+        "apf_not_found": ERROR_WITH_ACCESS_POINT_POLICY,
+    }
+    env = Environment(loader=DictLoader(RESTError.templates | extended_templates))
+
     def __init__(self, *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "single_error")
         super().__init__(*args, **kwargs)
@@ -24,7 +31,6 @@ class AccessPointNotFound(S3ControlError):
     def __init__(self, name: str, **kwargs: Any):
         kwargs.setdefault("template", "ap_not_found")
         kwargs["name"] = name
-        self.templates["ap_not_found"] = ERROR_WITH_ACCESS_POINT_NAME
         super().__init__(
             "NoSuchAccessPoint", "The specified accesspoint does not exist", **kwargs
         )
@@ -36,7 +42,6 @@ class AccessPointPolicyNotFound(S3ControlError):
     def __init__(self, name: str, **kwargs: Any):
         kwargs.setdefault("template", "apf_not_found")
         kwargs["name"] = name
-        self.templates["apf_not_found"] = ERROR_WITH_ACCESS_POINT_POLICY
         super().__init__(
             "NoSuchAccessPointPolicy",
             "The specified accesspoint policy does not exist",

moto/sagemaker/exceptions.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from jinja2 import DictLoader, Environment
 from moto.core.exceptions import AWSError, JsonRESTError, RESTError
 
 ERROR_WITH_MODEL_NAME = """{% extends 'single_error' %}
@@ -8,9 +9,11 @@
 
 
 class SagemakerClientError(RESTError):
+    extended_templates = {"model_error": ERROR_WITH_MODEL_NAME}
+    env = Environment(loader=DictLoader(RESTError.templates | extended_templates))
+
     def __init__(self, *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "single_error")
-        self.templates["model_error"] = ERROR_WITH_MODEL_NAME
         super().__init__(*args, **kwargs)
 
 

moto/sdb/exceptions.py

@@ -2,6 +2,7 @@
 from typing import Any
 
 from moto.core.exceptions import RESTError
+from jinja2 import DictLoader, Environment
 
 SDB_ERROR = """<?xml version="1.0"?>
 <Response>
@@ -18,6 +19,8 @@
 
 class InvalidParameterError(RESTError):
     code = 400
+    extended_templates = {"sdb_error": SDB_ERROR}
+    env = Environment(loader=DictLoader(RESTError.templates | extended_templates))
 
     def __init__(self, **kwargs: Any):
         kwargs.setdefault("template", "sdb_error")
@@ -37,10 +40,11 @@ def __init__(self, domain_name: str):
 
 class UnknownDomainName(RESTError):
     code = 400
+    extended_templates = {"sdb_error": SDB_ERROR}
+    env = Environment(loader=DictLoader(RESTError.templates | extended_templates))
 
     def __init__(self, **kwargs: Any):
         kwargs.setdefault("template", "sdb_error")
-        self.templates["sdb_error"] = SDB_ERROR
         kwargs["error_type"] = "NoSuchDomain"
         kwargs["message"] = "The specified domain does not exist."
         super().__init__(**kwargs)