|
1 | 1 | import logging
|
2 |
| -from typing import Any, Callable, Dict, Optional |
| 2 | +from typing import Any, Callable, Dict, Optional, Type, TypeVar, Union |
3 | 3 |
|
4 | 4 | from ...middleware_factory import lambda_handler_decorator
|
5 | 5 | from ..typing import LambdaContext
|
|
10 | 10 | logger = logging.getLogger(__name__)
|
11 | 11 |
|
12 | 12 |
|
| 13 | +EventParserReturnType = TypeVar("EventParserReturnType") |
| 14 | + |
| 15 | + |
13 | 16 | @lambda_handler_decorator
|
14 | 17 | def event_parser(
|
15 |
| - handler: Callable[[Any, LambdaContext], Any], |
| 18 | + handler: Callable[[Any, LambdaContext], EventParserReturnType], |
16 | 19 | event: Dict[str, Any],
|
17 | 20 | context: LambdaContext,
|
18 |
| - model: Model, |
19 |
| - envelope: Optional[Envelope] = None, |
20 |
| -) -> Any: |
| 21 | + model: Type[Model], |
| 22 | + envelope: Optional[Union[Envelope, Type[Envelope]]] = None, |
| 23 | +) -> EventParserReturnType: |
21 | 24 | """Lambda handler decorator to parse & validate events using Pydantic models
|
22 | 25 |
|
23 | 26 | It requires a model that implements Pydantic BaseModel to parse & validate the event.
|
@@ -83,7 +86,9 @@ def handler(event: Order, context: LambdaContext):
|
83 | 86 | return handler(parsed_event, context)
|
84 | 87 |
|
85 | 88 |
|
86 |
| -def parse(event: Dict[str, Any], model: Model, envelope: Optional[Envelope] = None) -> Model: |
| 89 | +def parse( |
| 90 | + event: Dict[str, Any], model: Type[Model], envelope: Optional[Union[Envelope, Type[Envelope]]] = None |
| 91 | +) -> Model: |
87 | 92 | """Standalone function to parse & validate events using Pydantic models
|
88 | 93 |
|
89 | 94 | Typically used when you need fine-grained control over error handling compared to event_parser decorator.
|
|
0 commit comments