6
6
from datetime import datetime
7
7
from typing import Callable , Optional , TypeVar , Union , overload
8
8
9
+ from typing_extensions import ParamSpec
10
+
9
11
from samtranslator .metrics .metrics import DummyMetricsPublisher , Metrics
10
12
from samtranslator .model import Resource
11
13
12
14
LOG = logging .getLogger (__name__ )
13
15
16
+ _PT = ParamSpec ("_PT" ) # parameters
14
17
_RT = TypeVar ("_RT" ) # return value
15
18
16
19
@@ -81,18 +84,18 @@ def _send_cw_metric(prefix, name, execution_time_ms, func, args): # type: ignor
81
84
@overload
82
85
def cw_timer (
83
86
* , name : Optional [str ] = None , prefix : Optional [str ] = None
84
- ) -> Callable [[Callable [... , _RT ]], Callable [... , _RT ]]:
87
+ ) -> Callable [[Callable [_PT , _RT ]], Callable [_PT , _RT ]]:
85
88
...
86
89
87
90
88
91
@overload
89
- def cw_timer (_func : Callable [... , _RT ], name : Optional [str ] = None , prefix : Optional [str ] = None ) -> Callable [... , _RT ]:
92
+ def cw_timer (_func : Callable [_PT , _RT ], name : Optional [str ] = None , prefix : Optional [str ] = None ) -> Callable [_PT , _RT ]:
90
93
...
91
94
92
95
93
96
def cw_timer (
94
- _func : Optional [Callable [... , _RT ]] = None , name : Optional [str ] = None , prefix : Optional [str ] = None
95
- ) -> Union [Callable [... , _RT ], Callable [[Callable [... , _RT ]], Callable [... , _RT ]]]:
97
+ _func : Optional [Callable [_PT , _RT ]] = None , name : Optional [str ] = None , prefix : Optional [str ] = None
98
+ ) -> Union [Callable [_PT , _RT ], Callable [[Callable [_PT , _RT ]], Callable [_PT , _RT ]]]:
96
99
"""
97
100
A method decorator, that will calculate execution time of the decorated method, and store this information as a
98
101
metric in CloudWatch by calling the metrics singleton instance.
@@ -105,7 +108,7 @@ def cw_timer(
105
108
If prefix is defined, it will be added in the beginning of what is been generated above
106
109
"""
107
110
108
- def cw_timer_decorator (func : Callable [... , _RT ]) -> Callable [... , _RT ]:
111
+ def cw_timer_decorator (func : Callable [_PT , _RT ]) -> Callable [_PT , _RT ]:
109
112
@functools .wraps (func )
110
113
def wrapper_cw_timer (* args , ** kwargs ) -> _RT : # type: ignore[no-untyped-def]
111
114
start_time = datetime .now ()
0 commit comments