1
1
import functools
2
2
import inspect
3
3
import logging
4
- import sys
5
4
import os
6
- from typing import Any , Callable , Dict , Optional , Union , cast , overload
7
-
8
- if sys .version_info >= (3 , 8 ):
9
- from typing import Protocol
10
- else :
11
- from typing_extensions import Protocol
5
+ from typing import Any , Callable , Optional
12
6
13
7
from ..shared import constants
14
8
from ..shared .functions import resolve_truthy_env_var_choice
15
9
from ..tracing import Tracer
16
- from ..utilities .typing import LambdaContext
17
10
from .exceptions import MiddlewareInvalidArgumentError
18
11
19
12
logger = logging .getLogger (__name__ )
20
13
21
- # context: Any to avoid forcing users to type it as context: LambdaContext
22
- _Handler = Callable [[Any , LambdaContext ], Any ]
23
- _RawHandlerDecorator = Callable [[_Handler ], _Handler ]
24
-
25
-
26
- class _FactoryDecorator (Protocol ):
27
- # it'd be better for this to be using ParamSpec (available from 3.10)
28
- def __call__ (
29
- self , handler : _Handler , event : Dict [str , Any ], context : LambdaContext , ** kwargs : Any
30
- ) -> _RawHandlerDecorator :
31
- ...
32
-
33
-
34
- class _HandlerDecorator (Protocol ):
35
- @overload
36
- def __call__ (self , decorator : _Handler ) -> _Handler :
37
- ...
38
-
39
- @overload
40
- def __call__ (self , decorator : None = None , ** kwargs : Any ) -> _RawHandlerDecorator :
41
- ...
42
-
43
- def __call__ (self , decorator : Optional [_Handler ] = None , ** kwargs : Any ) -> Union [_Handler , _RawHandlerDecorator ]:
44
- ...
45
-
46
-
47
- @overload
48
- def lambda_handler_decorator (decorator : _FactoryDecorator ) -> _HandlerDecorator :
49
- ...
50
-
51
-
52
- @overload
53
- def lambda_handler_decorator (
54
- decorator : None = None , trace_execution : Optional [bool ] = None
55
- ) -> Callable [[_FactoryDecorator ], _HandlerDecorator ]:
56
- ...
57
-
58
14
59
- def lambda_handler_decorator (
60
- decorator : Optional [_FactoryDecorator ] = None , trace_execution : Optional [bool ] = None
61
- ) -> Union [_HandlerDecorator , Callable [[_FactoryDecorator ], _HandlerDecorator ]]:
15
+ # giving this an accurate return type is hard
16
+ def lambda_handler_decorator (decorator : Optional [Callable ] = None , trace_execution : Optional [bool ] = None ) -> Callable :
62
17
"""Decorator factory for decorating Lambda handlers.
63
18
64
19
You can use lambda_handler_decorator to create your own middlewares,
@@ -149,25 +104,19 @@ def lambda_handler(event, context):
149
104
"""
150
105
151
106
if decorator is None :
152
- return cast (
153
- Callable [[_FactoryDecorator ], _HandlerDecorator ],
154
- functools .partial (lambda_handler_decorator , trace_execution = trace_execution ),
155
- )
107
+ return functools .partial (lambda_handler_decorator , trace_execution = trace_execution )
156
108
157
109
trace_execution = resolve_truthy_env_var_choice (
158
110
env = os .getenv (constants .MIDDLEWARE_FACTORY_TRACE_ENV , "false" ), choice = trace_execution
159
111
)
160
112
161
113
@functools .wraps (decorator )
162
- def final_decorator (
163
- func : Optional [_RawHandlerDecorator ] = None , ** kwargs : Any
164
- ) -> Union [_Handler , _RawHandlerDecorator ]:
114
+ def final_decorator (func : Optional [Callable ] = None , ** kwargs : Any ):
165
115
# If called with kwargs return new func with kwargs
166
116
if func is None :
167
117
return functools .partial (final_decorator , ** kwargs )
168
118
169
119
if not inspect .isfunction (func ):
170
- assert decorator is not None
171
120
# @custom_middleware(True) vs @custom_middleware(log_event=True)
172
121
raise MiddlewareInvalidArgumentError (
173
122
f"Only keyword arguments is supported for middlewares: { decorator .__qualname__ } received { func } " # type: ignore # noqa: E501
@@ -190,4 +139,4 @@ def wrapper(event, context):
190
139
191
140
return wrapper
192
141
193
- return cast ( _HandlerDecorator , final_decorator )
142
+ return final_decorator
0 commit comments