|
1 | 1 | import logging |
2 | | -from typing import Any, Callable, Dict, Optional, Type, TypeVar, Union |
| 2 | +from typing import Any, Callable, Dict, Optional, Type, overload |
| 3 | + |
| 4 | +from aws_lambda_powertools.utilities.parser.types import EnvelopeModel, EventParserReturnType, Model |
3 | 5 |
|
4 | 6 | from ...middleware_factory import lambda_handler_decorator |
5 | 7 | from ..typing import LambdaContext |
6 | 8 | from .envelopes.base import Envelope |
7 | 9 | from .exceptions import InvalidEnvelopeError, InvalidModelTypeError |
8 | | -from .types import Model |
9 | 10 |
|
10 | 11 | logger = logging.getLogger(__name__) |
11 | 12 |
|
12 | 13 |
|
13 | | -EventParserReturnType = TypeVar("EventParserReturnType") |
14 | | - |
15 | | - |
16 | 14 | @lambda_handler_decorator |
17 | 15 | def event_parser( |
18 | 16 | handler: Callable[[Any, LambdaContext], EventParserReturnType], |
19 | 17 | event: Dict[str, Any], |
20 | 18 | context: LambdaContext, |
21 | 19 | model: Type[Model], |
22 | | - envelope: Optional[Union[Envelope, Type[Envelope]]] = None, |
| 20 | + envelope: Optional[Type[Envelope]] = None, |
23 | 21 | ) -> EventParserReturnType: |
24 | 22 | """Lambda handler decorator to parse & validate events using Pydantic models |
25 | 23 |
|
@@ -81,14 +79,22 @@ def handler(event: Order, context: LambdaContext): |
81 | 79 | InvalidEnvelopeError |
82 | 80 | When envelope given does not implement BaseEnvelope |
83 | 81 | """ |
84 | | - parsed_event = parse(event=event, model=model, envelope=envelope) |
| 82 | + parsed_event = parse(event=event, model=model, envelope=envelope) if envelope else parse(event=event, model=model) |
85 | 83 | logger.debug(f"Calling handler {handler.__name__}") |
86 | 84 | return handler(parsed_event, context) |
87 | 85 |
|
88 | 86 |
|
89 | | -def parse( |
90 | | - event: Dict[str, Any], model: Type[Model], envelope: Optional[Union[Envelope, Type[Envelope]]] = None |
91 | | -) -> Model: |
| 87 | +@overload |
| 88 | +def parse(event: Dict[str, Any], model: Type[Model]) -> Model: |
| 89 | + ... |
| 90 | + |
| 91 | + |
| 92 | +@overload |
| 93 | +def parse(event: Dict[str, Any], model: Type[Model], envelope: Type[Envelope]) -> EnvelopeModel: |
| 94 | + ... |
| 95 | + |
| 96 | + |
| 97 | +def parse(event: Dict[str, Any], model: Type[Model], envelope: Optional[Type[Envelope]] = None): |
92 | 98 | """Standalone function to parse & validate events using Pydantic models |
93 | 99 |
|
94 | 100 | Typically used when you need fine-grained control over error handling compared to event_parser decorator. |
|
0 commit comments