forked from aws/aws-sam-cli
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcontext.py
More file actions
262 lines (217 loc) · 8.31 KB
/
context.py
File metadata and controls
262 lines (217 loc) · 8.31 KB
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
"""
Context information passed to each CLI command
"""
import logging
import uuid
from typing import List, Optional, cast
import click
from rich.console import Console
from samcli.cli.formatters import RootCommandHelpTextFormatter
from samcli.commands.exceptions import AWSServiceClientError
from samcli.lib.utils.sam_logging import (
LAMBDA_BULDERS_LOGGER_NAME,
SAM_CLI_FORMATTER_WITH_TIMESTAMP,
SAM_CLI_LOGGER_NAME,
SamCliLogger,
)
class Context:
"""
Top level context object for the CLI. Exposes common functionality required by a CLI, including logging,
environment config parsing, debug logging etc.
This object is passed by Click to every command that adds the proper annotation.
Read this for more details on Click Context - http://click.pocoo.org/5/commands/#nested-handling-and-contexts
Each command gets its own context object, but linked to both parent and child command's context, like a Linked List.
This class itself does not rely on how Click works. It is just a plain old Python class that holds common
properties used by every CLI command.
"""
_session_id: str
formatter_class = RootCommandHelpTextFormatter
def __init__(self):
"""
Initialize the context with default values
"""
self._debug = False
self._aws_region = None
self._aws_profile = None
self._session_id = str(uuid.uuid4())
self._experimental = False
self._exception = None
self._console = Console()
@property
def console(self):
return self._console
@property
def exception(self):
return self._exception
@exception.setter
def exception(self, value: Exception):
"""
Save the exception to handler in the future
Parameter
---------
value: Exception
The exception to save for future handling
"""
self._exception = value
@property
def debug(self):
return self._debug
@debug.setter
def debug(self, value):
"""
Turn on debug logging if necessary.
:param value: Value of debug flag
"""
self._debug = value
if self._debug:
# Turn on debug logging and display timestamps
sam_cli_logger = logging.getLogger(SAM_CLI_LOGGER_NAME)
lambda_builders_logger = logging.getLogger(LAMBDA_BULDERS_LOGGER_NAME)
SamCliLogger.configure_logger(sam_cli_logger, SAM_CLI_FORMATTER_WITH_TIMESTAMP, logging.DEBUG)
SamCliLogger.configure_logger(lambda_builders_logger, SAM_CLI_FORMATTER_WITH_TIMESTAMP, logging.DEBUG)
@property
def region(self):
return self._aws_region
@region.setter
def region(self, value):
"""
Set AWS region
"""
self._aws_region = value
self._refresh_session()
@property
def profile(self):
return self._aws_profile
@profile.setter
def profile(self, value):
"""
Set AWS profile for credential resolution
"""
self._aws_profile = value
self._refresh_session()
@property
def session_id(self) -> str:
"""
Returns the ID of this command session. This is a randomly generated UUIDv4 which will not change until the
command terminates.
"""
return self._session_id
@property
def experimental(self):
return self._experimental
@experimental.setter
def experimental(self, value):
self._experimental = value
@property
def command_path(self):
"""
Returns the full path of the command as invoked ex: "sam local generate-event s3 put". Wrapper to
https://click.palletsprojects.com/en/7.x/api/#click.Context.command_path
Returns
-------
str
Full path of the command invoked
"""
# Uses Click's Core Context. Note, this is different from this class, also confusingly named `Context`.
# Click's Core Context object is the one that contains command path information.
click_core_ctx = click.get_current_context()
if click_core_ctx:
return click_core_ctx.command_path
return None
@property
def template_dict(self):
"""
Returns the template_dictionary from click context.
Returns
-------
dict
Template as dictionary
"""
click_core_ctx = click.get_current_context()
try:
if click_core_ctx:
return click_core_ctx.template_dict
except AttributeError:
return None
return None
@staticmethod
def get_current_context() -> Optional["Context"]:
"""
Get the current Context object from Click's context stacks. This method is safe to run within the
actual command's handler that has a ``@pass_context`` annotation. Outside of the handler, you run
the risk of creating a new Context object which is entirely different from the Context object used by your
command.
.. code:
@pass_context
def my_command_handler(ctx):
# You will get the right context from within the command handler. This will also work from any
# downstream method invoked as part of the handler.
this_context = Context.get_current_context()
assert ctx == this_context
Returns
-------
samcli.cli.context.Context
Instance of this object, if we are running in a Click command. None otherwise.
"""
# Click has the concept of Context stacks. Think of them as linked list containing custom objects that are
# automatically accessible at different levels. We start from the Core Click context and discover the
# SAM CLI command-specific Context object which contains values for global options used by all commands.
#
# https://click.palletsprojects.com/en/7.x/complex/#ensuring-object-creation
#
click_core_ctx = click.get_current_context()
if click_core_ctx:
return cast("Context", click_core_ctx.find_object(Context) or click_core_ctx.ensure_object(Context))
return None
def _refresh_session(self):
"""
Update boto3's default session by creating a new session based on values set in the context. Some properties of
the Boto3's session object are read-only. Therefore when Click parses new AWS session related properties (like
region & profile), it will call this method to create a new session with latest values for these properties.
"""
import boto3
from botocore import credentials, exceptions, session
try:
botocore_session = session.get_session()
boto3.setup_default_session(
botocore_session=botocore_session, region_name=self._aws_region, profile_name=self._aws_profile
)
# get botocore session and setup caching for MFA based credentials
botocore_session.get_component("credential_provider").get_provider(
"assume-role"
).cache = credentials.JSONFileCache()
except exceptions.ProfileNotFound as ex:
raise AWSServiceClientError(str(ex)) from ex
def get_cmd_names(cmd_name, ctx) -> List[str]:
"""
Given the click core context, return a list representing all the subcommands passed to the CLI
Parameters
----------
cmd_name : str
name of current command
ctx : click.Context
click context
Returns
-------
list(str)
List containing subcommand names. Ex: ["local", "start-api"]
"""
if not ctx:
return []
if ctx and not getattr(ctx, "parent", None):
return [ctx.info_name]
# Find parent of current context
_parent = ctx.parent
_cmd_names = []
# Need to find the total set of commands that current command is part of.
if cmd_name != ctx.info_name:
_cmd_names = [cmd_name]
_cmd_names.append(ctx.info_name)
# Go through all parents till a parent of a context exists.
while _parent.parent:
info_name = _parent.info_name
_cmd_names.append(info_name)
_parent = _parent.parent
# Make sure the output reads natural. Ex: ["local", "start-api"]
_cmd_names.reverse()
return _cmd_names