improv: refactor structure to fit with utilities

Author: heitorlessa

aws_lambda_powertools/utilities/validation/__init__.py

@@ -0,0 +1,6 @@
+"""Validation utility
+"""
+from .envelopes import DynamoDBEnvelope, EventBridgeEnvelope, UserEnvelope
+from .validator import validator
+
+__all__ = ["UserEnvelope", "DynamoDBEnvelope", "EventBridgeEnvelope", "validator"]

aws_lambda_powertools/utilities/validation/envelopes/__init__.py

@@ -0,0 +1,9 @@
+from .base import UserEnvelope
+from .dynamodb import DynamoDBEnvelope
+from .event_bridge import EventBridgeEnvelope
+
+__all__ = [
+    "UserEnvelope",
+    "DynamoDBEnvelope",
+    "EventBridgeEnvelope",
+]

aws_lambda_powertools/utilities/validation/envelopes/base.py

@@ -0,0 +1,38 @@
+import logging
+from abc import ABC, abstractmethod
+from typing import Any, Dict
+
+from pydantic import BaseModel, ValidationError
+
+logger = logging.getLogger(__name__)
+
+
+class BaseEnvelope(ABC):
+    def _parse_user_dict_schema(self, user_event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+        logger.debug("parsing user dictionary schema")
+        try:
+            return inbound_schema_model(**user_event)
+        except (ValidationError, TypeError):
+            logger.exception("Valdation exception while extracting user custom schema")
+            raise
+
+    def _parse_user_json_string_schema(self, user_event: str, inbound_schema_model: BaseModel) -> Any:
+        logger.debug("parsing user dictionary schema")
+        try:
+            return inbound_schema_model.parse_raw(user_event)
+        except (ValidationError, TypeError):
+            logger.exception("Valdation exception while extracting user custom schema")
+            raise
+
+    @abstractmethod
+    def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+        return NotImplemented
+
+
+class UserEnvelope(BaseEnvelope):
+    def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+        try:
+            return inbound_schema_model(**event)
+        except (ValidationError, TypeError):
+            logger.exception("Valdation exception received from input user custom envelopes event")
+            raise

aws_lambda_powertools/utilities/validation/envelopes/dynamodb.py

@@ -0,0 +1,32 @@
+import logging
+from typing import Any, Dict
+
+from pydantic import BaseModel, ValidationError
+
+from ..schemas import DynamoDBSchema
+from .base import BaseEnvelope
+
+logger = logging.getLogger(__name__)
+
+
+class DynamoDBEnvelope(BaseEnvelope):
+    def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+        try:
+            parsed_envelope = DynamoDBSchema(**event)
+        except (ValidationError, TypeError):
+            logger.exception("Valdation exception received from input dynamodb stream event")
+            raise
+        output = []
+        for record in parsed_envelope.Records:
+            parsed_new_image = (
+                {}
+                if not record.dynamodb.NewImage
+                else self._parse_user_dict_schema(record.dynamodb.NewImage, inbound_schema_model)
+            )  # noqa: E501
+            parsed_old_image = (
+                {}
+                if not record.dynamodb.OldImage
+                else self._parse_user_dict_schema(record.dynamodb.OldImage, inbound_schema_model)
+            )  # noqa: E501
+            output.append({"new": parsed_new_image, "old": parsed_old_image})
+        return output

aws_lambda_powertools/utilities/validation/envelopes/event_bridge.py

@@ -0,0 +1,19 @@
+import logging
+from typing import Any, Dict
+
+from pydantic import BaseModel, ValidationError
+
+from ..schemas import EventBridgeSchema
+from .base import BaseEnvelope
+
+logger = logging.getLogger(__name__)
+
+
+class EventBridgeEnvelope(BaseEnvelope):
+    def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+        try:
+            parsed_envelope = EventBridgeSchema(**event)
+        except (ValidationError, TypeError):
+            logger.exception("Valdation exception received from input eventbridge event")
+            raise
+        return self._parse_user_dict_schema(parsed_envelope.detail, inbound_schema_model)

aws_lambda_powertools/utilities/validation/schemas/__init__.py

@@ -0,0 +1,11 @@
+from .dynamodb import DynamoDBSchema
+from .event_bridge import EventBridgeSchema
+from .sns import SnsSchema
+from .sqs import SqsSchema
+
+__all__ = [
+    "DynamoDBSchema",
+    "EventBridgeSchema",
+    "SnsSchema",
+    "SqsSchema",
+]

aws_lambda_powertools/validation/schemas.py renamed to aws_lambda_powertools/utilities/validation/schemas/dynamodb.py

@@ -1,4 +1,4 @@
-from datetime import date, datetime
+from datetime import date
 from typing import Any, Dict, List, Optional
 
 from pydantic import BaseModel, root_validator
@@ -35,22 +35,3 @@ class DynamoRecordSchema(BaseModel):
 
 class DynamoDBSchema(BaseModel):
     Records: List[DynamoRecordSchema]
-
-
-class EventBridgeSchema(BaseModel):
-    version: str
-    id: str  # noqa: A003,VNE003
-    source: str
-    account: int
-    time: datetime
-    region: str
-    resources: List[str]
-    detail: Dict[str, Any]
-
-
-class SqsSchema(BaseModel):
-    todo: str
-
-
-class SnsSchema(BaseModel):
-    todo: str

aws_lambda_powertools/utilities/validation/schemas/event_bridge.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+from typing import Any, Dict, List
+
+from pydantic import BaseModel
+
+
+class EventBridgeSchema(BaseModel):
+    version: str
+    id: str  # noqa: A003,VNE003
+    source: str
+    account: int
+    time: datetime
+    region: str
+    resources: List[str]
+    detail: Dict[str, Any]

aws_lambda_powertools/utilities/validation/schemas/sns.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class SnsSchema(BaseModel):
+    todo: str

aws_lambda_powertools/utilities/validation/schemas/sqs.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class SqsSchema(BaseModel):
+    todo: str