feat: RFC: Validate incoming and outgoing events utility #95

Author: Ran Isenberg

README.md

@@ -13,6 +13,7 @@ A suite of utilities for AWS Lambda Functions that makes tracing with AWS X-Ray,
 * **[Logging](https://awslabs.github.io/aws-lambda-powertools-python/core/logger/)** - Structured logging made easier, and decorator to enrich structured logging with key Lambda context details
 * **[Metrics](https://awslabs.github.io/aws-lambda-powertools-python/core/metrics/)** - Custom Metrics created asynchronously via CloudWatch Embedded Metric Format (EMF)
 * **[Bring your own middleware](https://awslabs.github.io/aws-lambda-powertools-python/utilities/middleware_factory/)** - Decorator factory to create your own middleware to run logic before, and after each Lambda invocation
+* **[Validation](https://)** - 
 
 ### Installation
 

aws_lambda_powertools/validation/__init__.py

@@ -0,0 +1,2 @@
+"""Validation utility
+"""

aws_lambda_powertools/validation/schemas.py

@@ -0,0 +1,56 @@
+from datetime import date, datetime
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, root_validator
+from typing_extensions import Literal
+
+
+class DynamoScheme(BaseModel):
+    ApproximateCreationDateTime: date
+    Keys: Dict[Literal["id"], Dict[Literal["S"], str]]
+    NewImage: Optional[Dict[str, Any]] = {}
+    OldImage: Optional[Dict[str, Any]] = {}
+    SequenceNumber: str
+    SizeBytes: int
+    StreamViewType: Literal["NEW_AND_OLD_IMAGES", "KEYS_ONLY", "NEW_IMAGE", "OLD_IMAGE"]
+
+    @root_validator
+    def check_one_image_exists(cls, values):
+        newimg, oldimg = values.get("NewImage"), values.get("OldImage")
+        stream_type = values.get("StreamViewType")
+        if stream_type == "NEW_AND_OLD_IMAGES" and not newimg and not oldimg:
+            raise TypeError("DynamoDB streams schema failed validation, missing both new & old stream images")
+        return values
+
+
+class DynamoRecordSchema(BaseModel):
+    eventID: str
+    eventName: Literal["INSERT", "MODIFY", "REMOVE"]
+    eventVersion: float
+    eventSource: Literal["aws:dynamodb"]
+    awsRegion: str
+    eventSourceARN: str
+    dynamodb: DynamoScheme
+
+
+class DynamoStreamSchema(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/validation/validator.py

@@ -0,0 +1,126 @@
+import logging
+from abc import ABC, abstractmethod
+from typing import Any, Callable, Dict
+
+from pydantic import BaseModel, ValidationError
+
+from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
+from aws_lambda_powertools.validation.schemas import DynamoStreamSchema, EventBridgeSchema
+
+logger = logging.getLogger(__name__)
+
+
+class Envelope(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(Envelope):
+    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 envelope event")
+            raise
+
+
+class EventBridgeEnvelope(Envelope):
+    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)
+
+
+class DynamoEnvelope(Envelope):
+    def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+        try:
+            parsed_envelope = DynamoStreamSchema(**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
+
+
+@lambda_handler_decorator
+def validator(
+    handler: Callable[[Dict, Any], Any],
+    event: Dict[str, Any],
+    context: Dict[str, Any],
+    inbound_schema_model: BaseModel,
+    outbound_schema_model: BaseModel,
+    envelope: Envelope,
+) -> Any:
+    """Decorator to create validation for lambda handlers events - both inbound and outbound
+
+        As Lambda follows (event, context) signature we can remove some of the boilerplate
+        and also capture any exception any Lambda function throws or its response as metadata
+
+        Example
+        -------
+        **Lambda function using validation decorator**
+
+            @validator(inbound=inbound_schema_model, outbound=outbound_schema_model)
+            def handler(parsed_event_model, context):
+                ...
+
+        Parameters
+        ----------
+        todo add
+
+        Raises
+        ------
+        err
+            TypeError or pydantic.ValidationError or any exception raised by the lambda handler itself
+    """
+    lambda_handler_name = handler.__name__
+    logger.debug("Validating inbound schema")
+    parsed_event_model = envelope.parse(event, inbound_schema_model)
+    try:
+        logger.debug(f"Calling handler {lambda_handler_name}")
+        response = handler({"orig": event, "custom": parsed_event_model}, context)
+        logger.debug("Received lambda handler response successfully")
+        logger.debug(response)
+    except Exception:
+        logger.exception(f"Exception received from {lambda_handler_name}")
+        raise
+
+    try:
+        logger.debug("Validating outbound response schema")
+        outbound_schema_model(**response)
+    except (ValidationError, TypeError):
+        logger.exception(f"Validation exception received from {lambda_handler_name} response event")
+        raise
+    return response