Speed up exception handling by re-using Environments (#7143)

moto/cloudfront/exceptions.py

@@ -15,10 +15,11 @@ EXCEPTION_RESPONSE = """<?xml version="1.0"?>
 
 class CloudFrontException(RESTError):
     code = 400
+    extended_templates = {"cferror": EXCEPTION_RESPONSE}
+    env = RESTError.extended_environment(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

@@ -1,5 +1,5 @@
 import json
-from typing import Any, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 
 from jinja2 import DictLoader, Environment
 from werkzeug.exceptions import HTTPException
@@ -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
@@ -62,9 +63,8 @@ class RESTError(HTTPException):
         self.error_type = error_type
         self.message = message
 
-        if template in self.templates.keys():
+        if template in self.env.list_templates():
-            env = Environment(loader=DictLoader(self.templates))
+            self.description: str = self.__class__.env.get_template(template).render(
-            self.description: str = env.get_template(template).render(
                 error_type=error_type,
                 message=message,
                 request_id_tag=self.request_id_tag_name,
@@ -95,6 +95,13 @@ class RESTError(HTTPException):
         err.code = self.code
         return err
 
+    @classmethod
+    def extended_environment(cls, extended_templates: Dict[str, str]) -> Environment:
+        # Can be simplified to cls.templates | extended_templates when we drop Python 3.8 support
+        # https://docs.python.org/3/library/stdtypes.html#mapping-types-dict
+        templates = dict(cls.templates.items() | extended_templates.items())
+        return Environment(loader=DictLoader(templates))
+
 
 class DryRunClientError(RESTError):
     code = 412

moto/ec2/exceptions.py

@@ -22,10 +22,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 = RESTError.extended_environment(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

@@ -16,10 +16,11 @@ EXCEPTION_RESPONSE = """<?xml version="1.0"?>
 class ElastiCacheException(RESTError):
 
     code = 400
+    extended_templates = {"ecerror": EXCEPTION_RESPONSE}
+    env = RESTError.extended_environment(extended_templates)
 
     def __init__(self, code: str, message: str, **kwargs: Any):
         kwargs.setdefault("template", "ecerror")
-        self.templates["ecerror"] = EXCEPTION_RESPONSE
         super().__init__(code, message)
 
 

moto/elasticbeanstalk/exceptions.py

@@ -16,10 +16,11 @@ EXCEPTION_RESPONSE = """<?xml version="1.0"?>
 class ElasticBeanstalkException(RESTError):
 
     code = 400
+    extended_templates = {"ecerror": EXCEPTION_RESPONSE}
+    env = RESTError.extended_environment(extended_templates)
 
     def __init__(self, code: str, message: str, **kwargs: Any):
         kwargs.setdefault("template", "ecerror")
-        self.templates["ecerror"] = EXCEPTION_RESPONSE
         super().__init__(code, message)
 
 

moto/s3/exceptions.py

@@ -41,9 +41,19 @@ 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 = RESTError.extended_environment(extended_templates)
+
     def __init__(self, *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "single_error")
-        self.templates["bucket_error"] = ERROR_WITH_BUCKET_NAME
         super().__init__(*args, **kwargs)
 
 
@@ -54,7 +64,6 @@ class InvalidArgumentError(S3ClientError):
         kwargs.setdefault("template", "argument_error")
         kwargs["name"] = name
         kwargs["value"] = value
-        self.templates["argument_error"] = ERROR_WITH_ARGUMENT
         super().__init__("InvalidArgument", message, *args, **kwargs)
 
 
@@ -75,7 +84,6 @@ class BadRequest(S3ClientError):
 class BucketError(S3ClientError):
     def __init__(self, *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "bucket_error")
-        self.templates["bucket_error"] = ERROR_WITH_BUCKET_NAME
         super().__init__(*args, **kwargs)
 
 
@@ -84,7 +92,6 @@ class BucketAlreadyExists(BucketError):
 
     def __init__(self, *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "bucket_error")
-        self.templates["bucket_error"] = ERROR_WITH_BUCKET_NAME
         super().__init__(
             "BucketAlreadyExists",
             (
@@ -111,7 +118,6 @@ class MissingKey(S3ClientError):
 
     def __init__(self, **kwargs: Any):
         kwargs.setdefault("template", "key_error")
-        self.templates["key_error"] = ERROR_WITH_KEY_NAME
         super().__init__("NoSuchKey", "The specified key does not exist.", **kwargs)
 
 
@@ -129,7 +135,6 @@ class InvalidVersion(S3ClientError):
         kwargs.setdefault("template", "argument_error")
         kwargs["name"] = "versionId"
         kwargs["value"] = version_id
-        self.templates["argument_error"] = ERROR_WITH_ARGUMENT
         super().__init__(
             "InvalidArgument", "Invalid version id specified", *args, **kwargs
         )
@@ -436,7 +441,6 @@ class NoSuchUpload(S3ClientError):
     def __init__(self, upload_id: Union[int, str], *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "error_uploadid")
         kwargs["upload_id"] = upload_id
-        self.templates["error_uploadid"] = ERROR_WITH_UPLOADID
         super().__init__(
             "NoSuchUpload",
             "The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.",
@@ -450,7 +454,6 @@ class PreconditionFailed(S3ClientError):
 
     def __init__(self, failed_condition: str, **kwargs: Any):
         kwargs.setdefault("template", "condition_error")
-        self.templates["condition_error"] = ERROR_WITH_CONDITION_NAME
         super().__init__(
             "PreconditionFailed",
             "At least one of the pre-conditions you specified did not hold",
@@ -464,7 +467,6 @@ class InvalidRange(S3ClientError):
 
     def __init__(self, range_requested: str, actual_size: str, **kwargs: Any):
         kwargs.setdefault("template", "range_error")
-        self.templates["range_error"] = ERROR_WITH_RANGE
         super().__init__(
             "InvalidRange",
             "The requested range is not satisfiable",
@@ -488,7 +490,6 @@ class InvalidObjectState(BucketError):
 
     def __init__(self, storage_class: Optional[str], **kwargs: Any):
         kwargs.setdefault("template", "storage_error")
-        self.templates["storage_error"] = ERROR_WITH_STORAGE_CLASS
         super().__init__(
             error_type="InvalidObjectState",
             message="The operation is not valid for the object's storage class",

moto/s3control/exceptions.py

@@ -13,6 +13,12 @@ ERROR_WITH_ACCESS_POINT_POLICY = """{% extends 'wrapped_single_error' %}
 
 
 class S3ControlError(RESTError):
+    extended_templates = {
+        "ap_not_found": ERROR_WITH_ACCESS_POINT_NAME,
+        "apf_not_found": ERROR_WITH_ACCESS_POINT_POLICY,
+    }
+    env = RESTError.extended_environment(extended_templates)
+
     def __init__(self, *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "single_error")
         super().__init__(*args, **kwargs)
@@ -24,7 +30,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 +41,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

@@ -8,16 +8,20 @@ ERROR_WITH_MODEL_NAME = """{% extends 'single_error' %}
 
 
 class SagemakerClientError(RESTError):
+    extended_templates = {"model_error": ERROR_WITH_MODEL_NAME}
+    env = RESTError.extended_environment(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)
 
 
 class ModelError(RESTError):
+    extended_templates = {"model_error": ERROR_WITH_MODEL_NAME}
+    env = RESTError.extended_environment(extended_templates)
+
     def __init__(self, *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "model_error")
-        self.templates["model_error"] = ERROR_WITH_MODEL_NAME
         super().__init__(*args, **kwargs)
 
 

moto/sdb/exceptions.py

@@ -18,10 +18,11 @@ SDB_ERROR = """<?xml version="1.0"?>
 
 class InvalidParameterError(RESTError):
     code = 400
+    extended_templates = {"sdb_error": SDB_ERROR}
+    env = RESTError.extended_environment(extended_templates)
 
     def __init__(self, **kwargs: Any):
         kwargs.setdefault("template", "sdb_error")
-        self.templates["sdb_error"] = SDB_ERROR
         kwargs["error_type"] = "InvalidParameterValue"
         super().__init__(**kwargs)
 
@@ -37,10 +38,11 @@ class InvalidDomainName(InvalidParameterError):
 
 class UnknownDomainName(RESTError):
     code = 400
+    extended_templates = {"sdb_error": SDB_ERROR}
+    env = RESTError.extended_environment(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)