-
Notifications
You must be signed in to change notification settings - Fork 1.2k
/
experimental.py
273 lines (218 loc) · 8.67 KB
/
experimental.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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
"""Experimental flag"""
import logging
import sys
from dataclasses import dataclass
from functools import wraps
from typing import Dict, List, Optional
import click
from samcli.cli.context import Context
from samcli.cli.global_config import ConfigEntry, GlobalConfig
from samcli.commands._utils.parameterized_option import parameterized_option
from samcli.lib.utils.colors import Colored, Colors
LOG = logging.getLogger(__name__)
EXPERIMENTAL_PROMPT = """
This feature is currently in beta.
Visit the docs page to learn more about the AWS Beta terms https://aws.amazon.com/service-terms/.
Enter Y to proceed with the command, or enter N to cancel:
"""
EXPERIMENTAL_WARNING = """
Experimental features are enabled for this session.
Visit the docs page to learn more about the AWS Beta terms https://aws.amazon.com/service-terms/.
"""
EXPERIMENTAL_ENV_VAR_PREFIX = "SAM_CLI_BETA_"
@dataclass(frozen=True, eq=True)
class ExperimentalEntry(ConfigEntry):
"""Child data class of ConfigEntry that enforces
config_key and env_var_key to be not None"""
config_key: str
env_var_key: str
persistent: bool = False
class ExperimentalFlag:
"""Class for storing all experimental related ConfigEntries"""
All = ExperimentalEntry("experimentalAll", EXPERIMENTAL_ENV_VAR_PREFIX + "FEATURES")
BuildPerformance = ExperimentalEntry(
"experimentalBuildPerformance", EXPERIMENTAL_ENV_VAR_PREFIX + "BUILD_PERFORMANCE"
)
IaCsSupport = {
"terraform": ExperimentalEntry(
"experimentalTerraformSupport", EXPERIMENTAL_ENV_VAR_PREFIX + "TERRAFORM_SUPPORT"
)
}
RustCargoLambda = ExperimentalEntry("experimentalCargoLambda", EXPERIMENTAL_ENV_VAR_PREFIX + "RUST_CARGO_LAMBDA")
def is_experimental_enabled(config_entry: ExperimentalEntry) -> bool:
"""Whether a given experimental flag is enabled or not.
If experimentalAll is set to True, then it will always return True.
Parameters
----------
config_entry : ExperimentalEntry
Experimental flag ExperimentalEntry
Returns
-------
bool
Whether the experimental flag is enabled or not.
"""
gc = GlobalConfig()
enabled = gc.get_value(config_entry, default=False, value_type=bool, is_flag=True)
if not enabled:
enabled = gc.get_value(ExperimentalFlag.All, default=False, value_type=bool, is_flag=True)
return enabled
def set_experimental(config_entry: ExperimentalEntry = ExperimentalFlag.All, enabled: bool = True) -> None:
"""Set the experimental flag to enabled or disabled.
Parameters
----------
config_entry : ExperimentalEntry, optional
Flag to be set, by default ExperimentalFlag.All
enabled : bool, optional
Enabled or disabled, by default True
"""
gc = GlobalConfig()
gc.set_value(config_entry, enabled, is_flag=True, flush=False)
def get_all_experimental() -> List[ExperimentalEntry]:
"""
Returns
-------
List[ExperimentalEntry]
List all experimental flags in the ExperimentalFlag class.
"""
all_experimental_flags = []
for name in dir(ExperimentalFlag):
if name.startswith("__"):
continue
value = getattr(ExperimentalFlag, name)
if isinstance(value, ExperimentalEntry):
all_experimental_flags.append(value)
elif isinstance(value, dict):
all_experimental_flags += value.values()
return all_experimental_flags
def get_all_experimental_env_vars() -> List[str]:
"""
Returns
-------
List[str]
List all env var names of experimental flags
"""
flags = get_all_experimental()
return [flag.env_var_key for flag in flags]
def get_all_experimental_statues() -> Dict[str, bool]:
"""Get statues of all experimental flags in a dictionary.
Returns
-------
Dict[str, bool]
Dictionary with key as configuration value and value as enabled or disabled.
"""
return {entry.config_key: is_experimental_enabled(entry) for entry in get_all_experimental() if entry.config_key}
def get_enabled_experimental_flags() -> List[str]:
"""
Returns a list of string, which contains enabled experimental flags for current session
Returns
-------
List[str]
List of strings which contains all enabled experimental flag names
"""
enabled_experimentals = []
for experimental_key, status in get_all_experimental_statues().items():
if status:
enabled_experimentals.append(experimental_key)
return enabled_experimentals
def disable_all_experimental():
"""Turn off all experimental flags in the ExperimentalFlag class."""
for entry in get_all_experimental():
set_experimental(entry, False)
def update_experimental_context(show_warning=True):
"""Set experimental for the current click context.
Parameters
----------
show_warning : bool, optional
Should warning be shown, by default True
"""
if not Context.get_current_context().experimental:
Context.get_current_context().experimental = True
if show_warning:
LOG.warning(Colored().color_log(EXPERIMENTAL_WARNING, color=Colors.WARNING), extra=dict(markup=True))
def _experimental_option_callback(ctx, param, enabled: Optional[bool]):
"""Click parameter callback for --beta-features or --no-beta-features.
If neither is specified, enabled will be None.
If --beta-features is set, enabled will be True,
we should turn on all experimental flags.
If --no-beta-features is set, enabled will be False,
we should turn off all experimental flags, overriding existing env vars.
"""
if enabled is None:
return
if enabled:
set_experimental(ExperimentalFlag.All, True)
update_experimental_context()
else:
disable_all_experimental()
def experimental_click_option(default: Optional[bool]):
return click.option(
"--beta-features/--no-beta-features",
default=default,
required=False,
is_flag=True,
expose_value=False,
callback=_experimental_option_callback,
help="Enable/Disable beta features.",
)
@parameterized_option
def experimental(f, default: Optional[bool] = None):
"""Decorator for adding --beta-features and --no-beta-features click options to a command."""
return experimental_click_option(default)(f)
@parameterized_option
def force_experimental(
f, config_entry: ExperimentalEntry = ExperimentalFlag.All, prompt=EXPERIMENTAL_PROMPT, default=None
):
"""Decorator for adding --beta-features and --no-beta-features click options to a command.
If experimental flag env var or --beta-features flag is not specified, this will then
prompt the user for confirmation.
The program will exit if confirmation is denied.
"""
def wrap(func):
@wraps(func)
def wrapped_func(*args, **kwargs):
if not prompt_experimental(config_entry=config_entry, prompt=prompt):
sys.exit(1)
return func(*args, **kwargs)
return wrapped_func
return experimental_click_option(default)(wrap(f))
@parameterized_option
def force_experimental_option(
f, option: str, config_entry: ExperimentalEntry = ExperimentalFlag.All, prompt=EXPERIMENTAL_PROMPT
):
"""Decorator for making a specific option to be experimental.
A prompt will be shown if experimental is not enabled and the option is specified.
"""
def wrap(func):
@wraps(func)
def wrapped_func(*args, **kwargs):
if kwargs[option]:
if not prompt_experimental(config_entry=config_entry, prompt=prompt):
sys.exit(1)
return func(*args, **kwargs)
return wrapped_func
return wrap(f)
def prompt_experimental(
config_entry: ExperimentalEntry = ExperimentalFlag.All, prompt: str = EXPERIMENTAL_PROMPT
) -> bool:
"""Prompt the user for experimental features.
If the corresponding experimental flag is already specified, the prompt will be skipped.
If confirmation is granted, the corresponding experimental flag env var will be set.
Parameters
----------
config_entry : ExperimentalEntry, optional
Which experimental flag should be set, by default ExperimentalFlag.All
prompt : str, optional
Text to be shown in the prompt, by default EXPERIMENTAL_PROMPT
Returns
-------
bool
Whether user have accepted the experimental feature.
"""
if is_experimental_enabled(config_entry):
update_experimental_context()
return True
confirmed = click.confirm(Colored().yellow(prompt), default=False)
if confirmed:
set_experimental(config_entry=config_entry, enabled=True)
update_experimental_context()
return confirmed