1
1
import logging
2
- from typing import Any , Callable , Dict , Optional
2
+ from typing import Any , Callable , Dict , Optional , Type , TypeVar
3
3
4
4
from pydantic import BaseModel
5
5
8
8
from .envelopes .base import BaseEnvelope
9
9
from .exceptions import InvalidEnvelopeError , InvalidModelTypeError
10
10
11
+ Model = TypeVar ("Model" , bound = BaseModel )
12
+ Envelope = TypeVar ("Envelope" , bound = BaseEnvelope )
11
13
logger = logging .getLogger (__name__ )
12
14
13
15
@@ -16,8 +18,8 @@ def event_parser(
16
18
handler : Callable [[Dict , Any ], Any ],
17
19
event : Dict [str , Any ],
18
20
context : LambdaContext ,
19
- model : BaseModel ,
20
- envelope : Optional [BaseEnvelope ] = None ,
21
+ model : Type [ Model ] ,
22
+ envelope : Optional [Type [ Envelope ] ] = None ,
21
23
) -> Any :
22
24
"""Lambda handler decorator to parse & validate events using Pydantic models
23
25
@@ -84,7 +86,7 @@ def handler(event: Order, context: LambdaContext):
84
86
return handler (parsed_event , context )
85
87
86
88
87
- def parse (event : Dict [str , Any ], model : BaseModel , envelope : Optional [BaseEnvelope ] = None ) -> Any :
89
+ def parse (event : Dict [str , Any ], model : Type [ Model ] , envelope : Optional [Type [ Envelope ] ] = None ) -> Any :
88
90
"""Standalone function to parse & validate events using Pydantic models
89
91
90
92
Typically used when you need fine-grained control over error handling compared to event_parser decorator.
@@ -152,4 +154,4 @@ def handler(event: Order, context: LambdaContext):
152
154
153
155
return model .parse_obj (event )
154
156
except AttributeError :
155
- raise InvalidModelTypeError ("Input model must implement BaseModel" )
157
+ raise InvalidModelTypeError (f "Input model must implement BaseModel, model= { model } " )
0 commit comments