forked from aws-powertools/powertools-lambda-python
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathfunctions.py
123 lines (89 loc) · 3.38 KB
/
functions.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
import base64
import itertools
import logging
import os
import warnings
from binascii import Error as BinAsciiError
from typing import Dict, Generator, Optional, Union, overload
from aws_lambda_powertools.shared import constants
logger = logging.getLogger(__name__)
def strtobool(value: str) -> bool:
"""Convert a string representation of truth to True or False.
True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values
are 'n', 'no', 'f', 'false', 'off', and '0'. Raises ValueError if
'value' is anything else.
> note:: Copied from distutils.util.
"""
value = value.lower()
if value in ("1", "y", "yes", "t", "true", "on"):
return True
if value in ("0", "n", "no", "f", "false", "off"):
return False
raise ValueError(f"invalid truth value {value!r}")
def resolve_truthy_env_var_choice(env: str, choice: Optional[bool] = None) -> bool:
"""Pick explicit choice over truthy env value, if available, otherwise return truthy env value
NOTE: Environment variable should be resolved by the caller.
Parameters
----------
env : str
environment variable actual value
choice : bool
explicit choice
Returns
-------
choice : str
resolved choice as either bool or environment value
"""
return choice if choice is not None else strtobool(env)
@overload
def resolve_env_var_choice(env: Optional[str], choice: float) -> float:
...
@overload
def resolve_env_var_choice(env: Optional[str], choice: str) -> str:
...
@overload
def resolve_env_var_choice(env: Optional[str], choice: Optional[str]) -> str:
...
def resolve_env_var_choice(
env: Optional[str] = None, choice: Optional[Union[str, float]] = None
) -> Optional[Union[str, float]]:
"""Pick explicit choice over env, if available, otherwise return env value received
NOTE: Environment variable should be resolved by the caller.
Parameters
----------
env : str, Optional
environment variable actual value
choice : str|float, optional
explicit choice
Returns
-------
choice : str, Optional
resolved choice as either bool or environment value
"""
return choice if choice is not None else env
def base64_decode(value: str) -> bytes:
try:
logger.debug("Decoding base64 record item before parsing")
return base64.b64decode(value)
except (BinAsciiError, TypeError):
raise ValueError("base64 decode failed")
def bytes_to_string(value: bytes) -> str:
try:
return value.decode("utf-8")
except (BinAsciiError, TypeError):
raise ValueError("base64 UTF-8 decode failed")
def powertools_dev_is_set() -> bool:
is_on = strtobool(os.getenv(constants.POWERTOOLS_DEV_ENV, "0"))
if is_on:
warnings.warn("POWERTOOLS_DEV environment variable is enabled. Increasing verbosity across utilities.")
return True
return False
def powertools_debug_is_set() -> bool:
is_on = strtobool(os.getenv(constants.POWERTOOLS_DEBUG_ENV, "0"))
if is_on:
warnings.warn("POWERTOOLS_DEBUG environment variable is enabled. Setting logging level to DEBUG.")
return True
return False
def slice_dictionary(data: Dict, chunk_size: int) -> Generator[Dict, None, None]:
for _ in range(0, len(data), chunk_size):
yield {dict_key: data[dict_key] for dict_key in itertools.islice(data, chunk_size)}