9
9
import os
10
10
from abc import ABC , abstractmethod
11
11
from datetime import datetime , timedelta
12
- from typing import (
13
- Any ,
14
- Callable ,
15
- Dict ,
16
- NamedTuple ,
17
- Optional ,
18
- Tuple ,
19
- Union ,
20
- cast ,
21
- overload ,
22
- )
12
+ from typing import TYPE_CHECKING , Any , Callable , NamedTuple , cast , overload
23
13
24
14
from aws_lambda_powertools .shared import constants , user_agent
25
15
from aws_lambda_powertools .shared .functions import resolve_max_age
26
- from aws_lambda_powertools .utilities .parameters .types import TransformOptions
27
16
28
17
from .exceptions import GetParameterError , TransformParameterError
29
18
19
+ if TYPE_CHECKING :
20
+ from aws_lambda_powertools .utilities .parameters .types import TransformOptions
21
+
30
22
DEFAULT_MAX_AGE_SECS = "300"
31
23
32
24
# These providers will be dynamically initialized on first use of the helper functions
33
- DEFAULT_PROVIDERS : Dict [str , Any ] = {}
25
+ DEFAULT_PROVIDERS : dict [str , Any ] = {}
34
26
TRANSFORM_METHOD_JSON = "json"
35
27
TRANSFORM_METHOD_BINARY = "binary"
36
28
SUPPORTED_TRANSFORM_METHODS = [TRANSFORM_METHOD_JSON , TRANSFORM_METHOD_BINARY ]
45
37
46
38
47
39
class ExpirableValue (NamedTuple ):
48
- value : str | bytes | Dict [str , Any ]
40
+ value : str | bytes | dict [str , Any ]
49
41
ttl : datetime
50
42
51
43
@@ -54,7 +46,7 @@ class BaseProvider(ABC):
54
46
Abstract Base Class for Parameter providers
55
47
"""
56
48
57
- store : Dict [ Tuple , ExpirableValue ]
49
+ store : dict [ tuple , ExpirableValue ]
58
50
59
51
def __init__ (self , * , client = None , resource = None ):
60
52
"""
@@ -65,19 +57,19 @@ def __init__(self, *, client=None, resource=None):
65
57
if resource is not None :
66
58
user_agent .register_feature_to_resource (resource = resource , feature = "parameters" )
67
59
68
- self .store : Dict [ Tuple , ExpirableValue ] = {}
60
+ self .store : dict [ tuple , ExpirableValue ] = {}
69
61
70
- def has_not_expired_in_cache (self , key : Tuple ) -> bool :
62
+ def has_not_expired_in_cache (self , key : tuple ) -> bool :
71
63
return key in self .store and self .store [key ].ttl >= datetime .now ()
72
64
73
65
def get (
74
66
self ,
75
67
name : str ,
76
- max_age : Optional [ int ] = None ,
68
+ max_age : int | None = None ,
77
69
transform : TransformOptions = None ,
78
70
force_fetch : bool = False ,
79
71
** sdk_options ,
80
- ) -> Optional [ Union [ str , dict , bytes ]] :
72
+ ) -> str | bytes | dict | None :
81
73
"""
82
74
Retrieve a parameter value or return the cached value
83
75
@@ -114,7 +106,7 @@ def get(
114
106
# of supported transform is small and the probability that a given
115
107
# parameter will always be used in a specific transform, this should be
116
108
# an acceptable tradeoff.
117
- value : Optional [ Union [ str , bytes , dict ]] = None
109
+ value : str | bytes | dict | None = None
118
110
key = self ._build_cache_key (name = name , transform = transform )
119
111
120
112
# If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS
@@ -139,7 +131,7 @@ def get(
139
131
return value
140
132
141
133
@abstractmethod
142
- def _get (self , name : str , ** sdk_options ) -> Union [ str , bytes , Dict [str , Any ] ]:
134
+ def _get (self , name : str , ** sdk_options ) -> str | bytes | dict [str , Any ]:
143
135
"""
144
136
Retrieve parameter value from the underlying parameter store
145
137
"""
@@ -154,12 +146,12 @@ def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs):
154
146
def get_multiple (
155
147
self ,
156
148
path : str ,
157
- max_age : Optional [ int ] = None ,
149
+ max_age : int | None = None ,
158
150
transform : TransformOptions = None ,
159
151
raise_on_transform_error : bool = False ,
160
152
force_fetch : bool = False ,
161
153
** sdk_options ,
162
- ) -> Union [ Dict [ str , str ], Dict [str , dict ], Dict [str , bytes ] ]:
154
+ ) -> dict [ str , str ] | dict [str , bytes ] | dict [str , dict ]:
163
155
"""
164
156
Retrieve multiple parameters based on a path prefix
165
157
@@ -211,7 +203,7 @@ def get_multiple(
211
203
return values
212
204
213
205
@abstractmethod
214
- def _get_multiple (self , path : str , ** sdk_options ) -> Dict [str , str ]:
206
+ def _get_multiple (self , path : str , ** sdk_options ) -> dict [str , str ]:
215
207
"""
216
208
Retrieve multiple parameter values from the underlying parameter store
217
209
"""
@@ -220,10 +212,10 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
220
212
def clear_cache (self ):
221
213
self .store .clear ()
222
214
223
- def fetch_from_cache (self , key : Tuple ):
215
+ def fetch_from_cache (self , key : tuple ):
224
216
return self .store [key ].value if key in self .store else {}
225
217
226
- def add_to_cache (self , key : Tuple , value : Any , max_age : int ):
218
+ def add_to_cache (self , key : tuple , value : Any , max_age : int ):
227
219
if max_age <= 0 :
228
220
return
229
221
@@ -248,7 +240,7 @@ def _build_cache_key(
248
240
249
241
Returns
250
242
-------
251
- Tuple [str, TransformOptions, bool]
243
+ tuple [str, TransformOptions, bool]
252
244
Cache key
253
245
"""
254
246
return (name , transform , is_nested )
@@ -294,28 +286,28 @@ def get_transform_method(value: str, transform: TransformOptions = None) -> Call
294
286
295
287
@overload
296
288
def transform_value (
297
- value : Dict [str , Any ],
289
+ value : dict [str , Any ],
298
290
transform : TransformOptions ,
299
291
raise_on_transform_error : bool = False ,
300
292
key : str = "" ,
301
- ) -> Dict [str , Any ]: ...
293
+ ) -> dict [str , Any ]: ...
302
294
303
295
304
296
@overload
305
297
def transform_value (
306
- value : Union [ str , bytes , Dict [str , Any ] ],
298
+ value : str | bytes | dict [str , Any ],
307
299
transform : TransformOptions ,
308
300
raise_on_transform_error : bool = False ,
309
301
key : str = "" ,
310
- ) -> Optional [ Union [ str , bytes , Dict [str , Any ]]] : ...
302
+ ) -> str | bytes | dict [str , Any ] | None : ...
311
303
312
304
313
305
def transform_value (
314
- value : Union [ str , bytes , Dict [str , Any ] ],
306
+ value : str | bytes | dict [str , Any ],
315
307
transform : TransformOptions ,
316
308
raise_on_transform_error : bool = True ,
317
309
key : str = "" ,
318
- ) -> Optional [ Union [ str , bytes , Dict [str , Any ]]] :
310
+ ) -> str | bytes | dict [str , Any ] | None :
319
311
"""
320
312
Transform a value using one of the available options.
321
313
@@ -348,7 +340,7 @@ def transform_value(
348
340
# where one of the keys might fail during transform, e.g. `{"a": "valid", "b": "{"}`
349
341
# expected: `{"a": "valid", "b": None}`
350
342
351
- transformed_values : Dict [str , Any ] = {}
343
+ transformed_values : dict [str , Any ] = {}
352
344
for dict_key , dict_value in value .items ():
353
345
transform_method = get_transform_method (value = dict_key , transform = transform )
354
346
try :
0 commit comments