feat: add standalone parse function

Author: heitorlessa

aws_lambda_powertools/utilities/parser/envelopes/base.py

@@ -4,7 +4,7 @@
 
 from pydantic import BaseModel, ValidationError
 
-from ..exceptions import InvalidEnvelopeError, SchemaValidationError
+from ..exceptions import SchemaValidationError
 
 logger = logging.getLogger(__name__)
 
@@ -29,11 +29,3 @@ def _parse(event: Union[Dict[str, Any], str], schema: BaseModel) -> Any:
     @abstractmethod
     def parse(self, event: Dict[str, Any], schema: BaseModel):
         return NotImplemented  # pragma: no cover
-
-
-def parse_envelope(event: Dict[str, Any], envelope: BaseEnvelope, schema: BaseModel):
-    try:
-        logger.debug(f"Parsing and validating event schema, envelope={envelope}")
-        return envelope().parse(event=event, schema=schema)
-    except (TypeError, AttributeError):
-        raise InvalidEnvelopeError(f"envelope must be a callable and instance of BaseEnvelope, envelope={envelope}")

aws_lambda_powertools/utilities/parser/parser.py

@@ -5,8 +5,8 @@
 
 from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
 
-from .envelopes.base import BaseEnvelope, parse_envelope
-from .exceptions import InvalidSchemaTypeError, SchemaValidationError
+from .envelopes.base import BaseEnvelope
+from .exceptions import InvalidEnvelopeError, InvalidSchemaTypeError, SchemaValidationError
 
 logger = logging.getLogger(__name__)
 
@@ -19,53 +19,96 @@ def parser(
     schema: BaseModel,
     envelope: Optional[BaseEnvelope] = None,
 ) -> Any:
-    # noinspection SpellCheckingInspection,SpellCheckingInspection
     """Decorator to conduct advanced parsing & validation for lambda handlers events
 
-            As Lambda follows (event, context) signature we can remove some of the boilerplate
-            and also capture any exception any Lambda function throws as metadata.
-            event will be the parsed and passed as a BaseModel pydantic class of the input type "schema"
-            to the lambda handler.
-            event will be extracted from the envelope in case envelope is not None.
-            In case envelope is None, the complete event is parsed to match the schema parameter BaseModel definition.
-            In case envelope is not None, first the event is parsed as the envelope's schema definition, and the user
-            message is extracted and parsed again as the schema parameter's definition.
-
-            Example
-            -------
-            **Lambda function using validation decorator**
-
-                @parser(schema=MyBusiness, envelope=envelopes.EVENTBRIDGE)
-                def handler(event: MyBusiness , context: LambdaContext):
-                    ...
-
-            Parameters
-            ----------
-            handler:  input for lambda_handler_decorator, wraps the handler lambda
-            event:    AWS event dictionary
-            context:  AWS lambda context
-            schema:   pydantic BaseModel class. This is the user data schema that will replace the event.
-                      event parameter will be parsed and a new schema object will be created from it.
-            envelope: what envelope to extract the schema from, can be any AWS service that is currently
-                      supported in the envelopes module. Can be None.
-
-            Raises
-            ------
-            SchemaValidationError
-                When input event doesn't conform with schema provided
-            InvalidSchemaTypeError
-                When schema given does not implement BaseModel
-        """
-    if envelope is None:
-        try:
-            logger.debug("Parsing and validating event schema; no envelope used")
-            parsed_event = schema.parse_obj(event)
-        except (ValidationError, TypeError) as e:
-            raise SchemaValidationError("Input event doesn't conform with schema") from e
-        except AttributeError:
-            raise InvalidSchemaTypeError("Input schema must implement BaseModel")
-    else:
-        parsed_event = parse_envelope(event=event, envelope=envelope, schema=schema)
+    As Lambda follows (event, context) signature we can remove some of the boilerplate
+    and also capture any exception any Lambda function throws as metadata.
+    event will be the parsed and passed as a BaseModel Pydantic class of the input type "schema"
+    to the lambda handler.
+    event will be extracted from the envelope in case envelope is not None.
+    In case envelope is None, the complete event is parsed to match the schema parameter BaseModel definition.
+    In case envelope is not None, first the event is parsed as the envelope's schema definition, and the user
+    message is extracted and parsed again as the schema parameter's definition.
+
+    Example
+    -------
+    **Lambda function using validation decorator**
+
+        @parser(schema=MyBusiness, envelope=envelopes.EVENTBRIDGE)
+        def handler(event: MyBusiness , context: LambdaContext):
+            ...
+
+    Parameters
+    ----------
+    handler:  input for lambda_handler_decorator, wraps the handler lambda
+    event:    AWS event dictionary
+    context:  AWS lambda context
+    schema:   pydantic BaseModel class. This is the user data schema that will replace the event.
+              event parameter will be parsed and a new schema object will be created from it.
+    envelope: what envelope to extract the schema from, can be any AWS service that is currently
+              supported in the envelopes module. Can be None.
 
+    Raises
+    ------
+    SchemaValidationError
+        When input event doesn't conform with schema provided
+    InvalidSchemaTypeError
+        When schema given does not implement BaseModel
+    """
+    parsed_event = parse(event=event, schema=schema, envelope=envelope)
     logger.debug(f"Calling handler {handler.__name__}")
     return handler(parsed_event, context)
+
+
+def parse(event: Dict[str, Any], schema: BaseModel, envelope: Optional[BaseEnvelope] = None) -> Any:
+    """
+    Standalone parse function to conduct advanced parsing & validation for lambda handlers events
+
+    As Lambda follows (event, context) signature we can remove some of the boilerplate
+    and also capture any exception any Lambda function throws as metadata.
+    event will be the parsed and passed as a BaseModel Pydantic class of the input type "schema"
+    to the lambda handler.
+    event will be extracted from the envelope in case envelope is not None.
+    In case envelope is None, the complete event is parsed to match the schema parameter BaseModel definition.
+    In case envelope is not None, first the event is parsed as the envelope's schema definition, and the user
+    message is extracted and parsed again as the schema parameter's definition.
+
+    Example
+    -------
+    **Lambda function using standalone parse**
+
+        def handler(event: MyBusiness , context: LambdaContext):
+            parse(event=event, schema=MyBusiness, envelope=envelopes.EVENTBRIDGE)
+            ...
+
+    Parameters
+    ----------
+    event:    AWS event dictionary
+    schema:   pydantic BaseModel class. This is the user data schema that will replace the event.
+              event parameter will be parsed and a new schema object will be created from it.
+    envelope: what envelope to extract the schema from, can be any AWS service that is currently
+              supported in the envelopes module. Can be None.
+
+    Raises
+    ------
+    SchemaValidationError
+        When input event doesn't conform with schema provided
+    InvalidSchemaTypeError
+        When schema given does not implement BaseModel
+
+    """
+    if envelope:
+        try:
+            logger.debug(f"Parsing and validating event schema, envelope={envelope}")
+            # noinspection PyCallingNonCallable
+            return envelope().parse(event=event, schema=schema)
+        except (TypeError, AttributeError):
+            raise InvalidEnvelopeError(f"envelope must be a callable and instance of BaseEnvelope, envelope={envelope}")
+
+    try:
+        logger.debug("Parsing and validating event schema; no envelope used")
+        return schema.parse_obj(event)
+    except (ValidationError, TypeError) as e:
+        raise SchemaValidationError("Input event doesn't conform with schema") from e
+    except AttributeError:
+        raise InvalidSchemaTypeError("Input schema must implement BaseModel")

tests/functional/parser/test_parser.py

@@ -16,14 +16,14 @@ def handle_no_envelope(event: Dict, _: LambdaContext):
         handle_no_envelope(event=invalid_value, context=LambdaContext())
 
 
-@pytest.mark.parametrize("invalid_envelope", [bool(), [], (), object])
-def test_parser_invalid_envelope_type(dummy_schema, invalid_envelope):
+@pytest.mark.parametrize("invalid_envelope", [True, ["dummy"], ("dummy"), object])
+def test_parser_invalid_envelope_type(dummy_event, dummy_schema, invalid_envelope):
     @parser(schema=dummy_schema, envelope=invalid_envelope)
     def handle_no_envelope(event: Dict, _: LambdaContext):
         return event
 
     with pytest.raises(exceptions.InvalidEnvelopeError):
-        handle_no_envelope(event={}, context=LambdaContext())
+        handle_no_envelope(event=dummy_event["payload"], context=LambdaContext())
 
 
 def test_parser_schema_with_envelope(dummy_event, dummy_schema, dummy_envelope):