2
2
import inspect
3
3
import logging
4
4
import os
5
- from typing import Any , Callable , Optional
5
+ from typing import Any , Callable , Optional , TypeVar , cast
6
6
7
7
from ..shared import constants
8
8
from ..shared .functions import resolve_truthy_env_var_choice
11
11
12
12
logger = logging .getLogger (__name__ )
13
13
14
+ FuncType = TypeVar ("FuncType" , bound = Callable [..., Any ])
15
+
14
16
15
17
# Maintenance: we can't yet provide an accurate return type without ParamSpec etc. see #1066
16
- def lambda_handler_decorator (decorator : Optional [Callable ] = None , trace_execution : Optional [bool ] = None ) -> Callable :
18
+ def lambda_handler_decorator (
19
+ decorator : Optional [Callable [..., Any ]] = None ,
20
+ trace_execution : Optional [bool ] = None ,
21
+ ) -> Callable [[Callable [..., Any ]], Callable [..., Any ]]:
17
22
"""Decorator factory for decorating Lambda handlers.
18
23
19
24
You can use lambda_handler_decorator to create your own middlewares,
@@ -34,7 +39,7 @@ def lambda_handler_decorator(decorator: Optional[Callable] = None, trace_executi
34
39
35
40
Parameters
36
41
----------
37
- decorator: Callable
42
+ decorator: Callable[..., Any]
38
43
Middleware to be wrapped by this factory
39
44
trace_execution: bool
40
45
Flag to explicitly enable trace execution for middlewares.\n
@@ -104,18 +109,21 @@ def lambda_handler(event, context):
104
109
"""
105
110
106
111
if decorator is None :
107
- return functools .partial (lambda_handler_decorator , trace_execution = trace_execution )
112
+ return cast (
113
+ Callable [[Callable [..., Any ]], Callable [..., Any ]],
114
+ functools .partial (lambda_handler_decorator , trace_execution = trace_execution ),
115
+ )
108
116
109
117
trace_execution = resolve_truthy_env_var_choice (
110
118
env = os .getenv (constants .MIDDLEWARE_FACTORY_TRACE_ENV , "false" ),
111
119
choice = trace_execution ,
112
120
)
113
121
114
122
@functools .wraps (decorator )
115
- def final_decorator (func : Optional [Callable ] = None , ** kwargs : Any ):
123
+ def final_decorator (func : Optional [FuncType ] = None , * args : Any , ** kwargs : Any ) -> FuncType :
116
124
# If called with kwargs return new func with kwargs
117
125
if func is None :
118
- return functools .partial (final_decorator , ** kwargs )
126
+ return cast ( FuncType , functools .partial (final_decorator , * args , ** kwargs ) )
119
127
120
128
if not inspect .isfunction (func ):
121
129
# @custom_middleware(True) vs @custom_middleware(log_event=True)
@@ -124,9 +132,11 @@ def final_decorator(func: Optional[Callable] = None, **kwargs: Any):
124
132
)
125
133
126
134
@functools .wraps (func )
127
- def wrapper (event , context , ** handler_kwargs ) :
135
+ def wrapper (* args : Any , ** kwargs : Any ) -> Any :
128
136
try :
129
- middleware = functools .partial (decorator , func , event , context , ** kwargs , ** handler_kwargs )
137
+ if decorator is None :
138
+ raise ValueError ("Decorator cannot be None" )
139
+ middleware = functools .partial (decorator , func , * args , ** kwargs )
130
140
if trace_execution :
131
141
tracer = Tracer (auto_patch = False )
132
142
with tracer .provider .in_subsegment (name = f"## { decorator .__qualname__ } " ):
@@ -135,9 +145,11 @@ def wrapper(event, context, **handler_kwargs):
135
145
response = middleware ()
136
146
return response
137
147
except Exception :
138
- logger .exception (f"Caught exception in { decorator .__qualname__ } " )
148
+ logger .exception (
149
+ f"Caught exception in { decorator .__qualname__ if decorator is not None else 'UnknownDecorator' } " ,
150
+ )
139
151
raise
140
152
141
- return wrapper
153
+ return cast ( FuncType , wrapper )
142
154
143
155
return final_decorator
0 commit comments