chore: Improve exceptions handling. Have exceptions grouped by attribute name and typed exceptions

Signed-off-by: Tudor Plugaru <[email protected]>

Author: PlugaruT

src/cloudevents/core/v1/event.py

@@ -13,10 +13,17 @@
 #    under the License.
 
 import re
+from collections import defaultdict
 from datetime import datetime
 from typing import Any, Final, Optional
 
-from cloudevents.core.v1.exceptions import CloudEventValidationError
+from cloudevents.core.v1.exceptions import (
+    BaseCloudEventException,
+    CloudEventValidationError,
+    CustomExtensionAttributeError,
+    InvalidAttributeTypeError,
+    MissingRequiredAttributeError,
+)
 
 REQUIRED_ATTRIBUTES: Final[list[str]] = ["id", "source", "type", "specversion"]
 OPTIONAL_ATTRIBUTES: Final[list[str]] = [
@@ -57,128 +64,157 @@ def _validate_attribute(attributes: dict[str, Any]) -> None:
 
         See https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md#required-attributes
         """
-        errors = {}
+        errors: dict[str, list] = defaultdict(list)
         errors.update(CloudEvent._validate_required_attributes(attributes))
-        errors.update(CloudEvent._validate_attribute_types(attributes))
         errors.update(CloudEvent._validate_optional_attributes(attributes))
         errors.update(CloudEvent._validate_extension_attributes(attributes))
         if errors:
-            raise CloudEventValidationError(errors)
+            raise CloudEventValidationError(dict(errors))
 
     @staticmethod
     def _validate_required_attributes(
         attributes: dict[str, Any],
-    ) -> dict[str, list[str]]:
-        """
-        Validates that all required attributes are present.
-
-        :param attributes: The attributes of the CloudEvent instance.
-        :return: A dictionary of validation error messages.
-        """
-        errors = {}
-        missing_attributes = [
-            attr for attr in REQUIRED_ATTRIBUTES if attr not in attributes
-        ]
-        if missing_attributes:
-            errors["required"] = [
-                f"Missing required attribute(s): {', '.join(missing_attributes)}"
-            ]
-        return errors
-
-    @staticmethod
-    def _validate_attribute_types(attributes: dict[str, Any]) -> dict[str, list[str]]:
+    ) -> dict[str, list[BaseCloudEventException]]:
         """
         Validates the types of the required attributes.
 
         :param attributes: The attributes of the CloudEvent instance.
         :return: A dictionary of validation error messages.
         """
-        errors = {}
-        type_errors = []
+        errors = defaultdict(list)
+
+        if "id" not in attributes:
+            errors["id"].append(MissingRequiredAttributeError(missing="id"))
         if attributes.get("id") is None:
-            type_errors.append("Attribute 'id' must not be None")
+            errors["id"].append(
+                InvalidAttributeTypeError("Attribute 'id' must not be None")
+            )
         if not isinstance(attributes.get("id"), str):
-            type_errors.append("Attribute 'id' must be a string")
+            errors["id"].append(
+                InvalidAttributeTypeError("Attribute 'id' must be a string")
+            )
+
+        if "source" not in attributes:
+            errors["source"].append(MissingRequiredAttributeError(missing="source"))
         if not isinstance(attributes.get("source"), str):
-            type_errors.append("Attribute 'source' must be a string")
+            errors["source"].append(
+                InvalidAttributeTypeError("Attribute 'source' must be a string")
+            )
+
+        if "type" not in attributes:
+            errors["type"].append(MissingRequiredAttributeError(missing="type"))
         if not isinstance(attributes.get("type"), str):
-            type_errors.append("Attribute 'type' must be a string")
+            errors["type"].append(
+                InvalidAttributeTypeError("Attribute 'type' must be a string")
+            )
+
+        if "specversion" not in attributes:
+            errors["specversion"].append(
+                MissingRequiredAttributeError(missing="specversion")
+            )
         if not isinstance(attributes.get("specversion"), str):
-            type_errors.append("Attribute 'specversion' must be a string")
+            errors["specversion"].append(
+                InvalidAttributeTypeError("Attribute 'specversion' must be a string")
+            )
         if attributes.get("specversion") != "1.0":
-            type_errors.append("Attribute 'specversion' must be '1.0'")
-        if type_errors:
-            errors["type"] = type_errors
+            errors["specversion"].append(
+                InvalidAttributeTypeError("Attribute 'specversion' must be '1.0'")
+            )
         return errors
 
     @staticmethod
     def _validate_optional_attributes(
         attributes: dict[str, Any],
-    ) -> dict[str, list[str]]:
+    ) -> dict[str, list[BaseCloudEventException]]:
         """
         Validates the types and values of the optional attributes.
 
         :param attributes: The attributes of the CloudEvent instance.
         :return: A dictionary of validation error messages.
         """
-        errors = {}
-        optional_errors = []
+        errors = defaultdict(list)
+
         if "time" in attributes:
             if not isinstance(attributes["time"], datetime):
-                optional_errors.append("Attribute 'time' must be a datetime object")
+                errors["time"].append(
+                    InvalidAttributeTypeError(
+                        "Attribute 'time' must be a datetime object"
+                    )
+                )
             if hasattr(attributes["time"], "tzinfo") and not attributes["time"].tzinfo:
-                optional_errors.append("Attribute 'time' must be timezone aware")
+                errors["time"].append(
+                    InvalidAttributeTypeError("Attribute 'time' must be timezone aware")
+                )
         if "subject" in attributes:
             if not isinstance(attributes["subject"], str):
-                optional_errors.append("Attribute 'subject' must be a string")
+                errors["subject"].append(
+                    InvalidAttributeTypeError("Attribute 'subject' must be a string")
+                )
             if not attributes["subject"]:
-                optional_errors.append("Attribute 'subject' must not be empty")
+                errors["subject"].append(
+                    InvalidAttributeTypeError("Attribute 'subject' must not be empty")
+                )
         if "datacontenttype" in attributes:
             if not isinstance(attributes["datacontenttype"], str):
-                optional_errors.append("Attribute 'datacontenttype' must be a string")
+                errors["datacontenttype"].append(
+                    InvalidAttributeTypeError(
+                        "Attribute 'datacontenttype' must be a string"
+                    )
+                )
             if not attributes["datacontenttype"]:
-                optional_errors.append("Attribute 'datacontenttype' must not be empty")
+                errors["datacontenttype"].append(
+                    InvalidAttributeTypeError(
+                        "Attribute 'datacontenttype' must not be empty"
+                    )
+                )
         if "dataschema" in attributes:
             if not isinstance(attributes["dataschema"], str):
-                optional_errors.append("Attribute 'dataschema' must be a string")
+                errors["dataschema"].append(
+                    InvalidAttributeTypeError("Attribute 'dataschema' must be a string")
+                )
             if not attributes["dataschema"]:
-                optional_errors.append("Attribute 'dataschema' must not be empty")
-        if optional_errors:
-            errors["optional"] = optional_errors
+                errors["dataschema"].append(
+                    InvalidAttributeTypeError(
+                        "Attribute 'dataschema' must not be empty"
+                    )
+                )
         return errors
 
     @staticmethod
     def _validate_extension_attributes(
         attributes: dict[str, Any],
-    ) -> dict[str, list[str]]:
+    ) -> dict[str, list[BaseCloudEventException]]:
         """
         Validates the extension attributes.
 
         :param attributes: The attributes of the CloudEvent instance.
         :return: A dictionary of validation error messages.
         """
-        errors = {}
-        extension_errors = []
+        errors = defaultdict(list)
         extension_attributes = [
             key
             for key in attributes.keys()
             if key not in REQUIRED_ATTRIBUTES and key not in OPTIONAL_ATTRIBUTES
         ]
         for extension_attribute in extension_attributes:
             if extension_attribute == "data":
-                extension_errors.append(
-                    "Extension attribute 'data' is reserved and must not be used"
+                errors[extension_attribute].append(
+                    CustomExtensionAttributeError(
+                        "Extension attribute 'data' is reserved and must not be used"
+                    )
                 )
             if not (1 <= len(extension_attribute) <= 20):
-                extension_errors.append(
-                    f"Extension attribute '{extension_attribute}' should be between 1 and 20 characters long"
+                errors[extension_attribute].append(
+                    CustomExtensionAttributeError(
+                        f"Extension attribute '{extension_attribute}' should be between 1 and 20 characters long"
+                    )
                 )
             if not re.match(r"^[a-z0-9]+$", extension_attribute):
-                extension_errors.append(
-                    f"Extension attribute '{extension_attribute}' should only contain lowercase letters and numbers"
+                errors[extension_attribute].append(
+                    CustomExtensionAttributeError(
+                        f"Extension attribute '{extension_attribute}' should only contain lowercase letters and numbers"
+                    )
                 )
-        if extension_errors:
-            errors["extensions"] = extension_errors
         return errors
 
     def get_id(self) -> str:

src/cloudevents/core/v1/exceptions.py

@@ -11,17 +11,46 @@
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    License for the specific language governing permissions and limitations
 #    under the License.
-class CloudEventValidationError(Exception):
+class BaseCloudEventException(Exception):
+    pass
+
+
+class CloudEventValidationError(BaseCloudEventException):
     """
-    Custom exception for validation errors.
+    Holds validation errors aggregated during a CloudEvent creation.
     """
 
-    def __init__(self, errors: dict[str, list[str]]) -> None:
+    def __init__(self, errors: dict[str, list[BaseCloudEventException]]) -> None:
+        """
+        :param errors: The errors gathered during the CloudEvent creation where key
+            is the name of the attribute and value is a list of errors related to that attribute.
+        """
         super().__init__("Validation errors occurred")
-        self.errors: dict[str, list[str]] = errors
+        self.errors: dict[str, list[BaseCloudEventException]] = errors
 
     def __str__(self) -> str:
         error_messages = [
-            f"{key}: {', '.join(value)}" for key, value in self.errors.items()
+            f"{key}: {', '.join(str(value))}" for key, value in self.errors.items()
         ]
         return f"{super().__str__()}: {', '.join(error_messages)}"
+
+
+class MissingRequiredAttributeError(BaseCloudEventException):
+    """
+    Exception for missing required attribute.
+    """
+
+    def __init__(self, missing: str) -> None:
+        super().__init__(f"Missing required attribute: '{missing}'")
+
+
+class CustomExtensionAttributeError(BaseCloudEventException):
+    """
+    Exception for invalid custom extension names.
+    """
+
+
+class InvalidAttributeTypeError(BaseCloudEventException):
+    """
+    Exception for invalid attribute type.
+    """