feat(parser): infer model from type hint (#3181)

* feat(parser): infer model from type hint if possible

* chore(parser): remove unnecessary branch

* Small changes

---------

Co-authored-by: Leandro Damascena <[email protected]>

Author: Tom01098

aws_lambda_powertools/utilities/parser/parser.py

@@ -1,4 +1,5 @@
 import logging
+import typing
 from typing import Any, Callable, Dict, Optional, Type, overload
 
 from aws_lambda_powertools.utilities.parser.compat import disable_pydantic_v2_warning
@@ -17,7 +18,7 @@ def event_parser(
     handler: Callable[[Any, LambdaContext], EventParserReturnType],
     event: Dict[str, Any],
     context: LambdaContext,
-    model: Type[Model],
+    model: Optional[Type[Model]] = None,
     envelope: Optional[Type[Envelope]] = None,
 ) -> EventParserReturnType:
     """Lambda handler decorator to parse & validate events using Pydantic models
@@ -76,10 +77,22 @@ def handler(event: Order, context: LambdaContext):
     ValidationError
         When input event does not conform with model provided
     InvalidModelTypeError
-        When model given does not implement BaseModel
+        When model given does not implement BaseModel or is not provided
     InvalidEnvelopeError
         When envelope given does not implement BaseEnvelope
     """
+
+    # The first parameter of a Lambda function is always the event
+    # This line get the model informed in the event_parser function
+    # or the first parameter of the function by using typing.get_type_hints
+    type_hints = typing.get_type_hints(handler)
+    model = model or (list(type_hints.values())[0] if type_hints else None)
+    if model is None:
+        raise InvalidModelTypeError(
+            "The model must be provided either as the `model` argument to `event_parser`"
+            "or as the type hint of `event` in the handler that it wraps",
+        )
+
     parsed_event = parse(event=event, model=model, envelope=envelope) if envelope else parse(event=event, model=model)
     logger.debug(f"Calling handler {handler.__name__}")
     return handler(parsed_event, context)

docs/utilities/parser.md

@@ -118,6 +118,12 @@ handler(event=payload, context=LambdaContext())
 handler(event=json.dumps(payload), context=LambdaContext()) # also works if event is a JSON string
 ```
 
+Alternatively, you can automatically extract the model from the `event` without the need to include the model parameter in the `event_parser` function.
+
+```python hl_lines="23 24"
+    --8<-- "examples/parser/src/using_the_model_from_event.py"
+```
+
 #### parse function
 
 Use this standalone function when you want more control over the data validation process, for example returning a 400 error for malformed payloads.

examples/parser/src/using_the_model_from_event.py

@@ -0,0 +1,27 @@
+import json
+
+from pydantic import BaseModel, validator
+
+from aws_lambda_powertools.utilities.parser import event_parser
+from aws_lambda_powertools.utilities.parser.models import APIGatewayProxyEventV2Model
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+
+class CancelOrder(BaseModel):
+    order_id: int
+    reason: str
+
+
+class CancelOrderModel(APIGatewayProxyEventV2Model):
+    body: CancelOrder  # type: ignore[assignment]
+
+    @validator("body", pre=True)
+    def transform_body_to_dict(cls, value: str):
+        return json.loads(value)
+
+
+@event_parser
+def handler(event: CancelOrderModel, context: LambdaContext):
+    cancel_order: CancelOrder = event.body
+
+    assert cancel_order.order_id is not None

tests/functional/parser/test_parser.py

@@ -93,3 +93,28 @@ def handle_no_envelope(event: Union[Dict, str], _: LambdaContext):
         return event
 
     handle_no_envelope(dummy_event, LambdaContext())
+
+
+def test_parser_event_with_type_hint(dummy_event, dummy_schema):
+    @event_parser
+    def handler(event: dummy_schema, _: LambdaContext):
+        assert event.message == "hello world"
+
+    handler(dummy_event["payload"], LambdaContext())
+
+
+def test_parser_event_without_type_hint(dummy_event, dummy_schema):
+    @event_parser
+    def handler(event, _):
+        assert event.message == "hello world"
+
+    with pytest.raises(exceptions.InvalidModelTypeError):
+        handler(dummy_event["payload"], LambdaContext())
+
+
+def test_parser_event_with_type_hint_and_non_default_argument(dummy_event, dummy_schema):
+    @event_parser
+    def handler(evt: dummy_schema, _: LambdaContext):
+        assert evt.message == "hello world"
+
+    handler(dummy_event["payload"], LambdaContext())