-
Notifications
You must be signed in to change notification settings - Fork 4.3k
/
Copy pathcommands.py
443 lines (377 loc) · 16.8 KB
/
commands.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
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
import logging
import os
from botocore import model
from botocore.compat import OrderedDict
from botocore.validate import validate_parameters
import awscli
from awscli.argparser import ArgTableArgParser
from awscli.argprocess import unpack_argument, unpack_cli_arg
from awscli.arguments import CustomArgument, create_argument_model_from_schema
from awscli.clidocs import OperationDocumentEventHandler
from awscli.clidriver import CLICommand
from awscli.bcdoc import docevents
from awscli.help import HelpCommand
from awscli.schema import SchemaTransformer
LOG = logging.getLogger(__name__)
_open = open
class _FromFile(object):
def __init__(self, *paths, **kwargs):
"""
``**kwargs`` can contain a ``root_module`` argument
that contains the root module where the file contents
should be searched. This is an optional argument, and if
no value is provided, will default to ``awscli``. This means
that by default we look for examples in the ``awscli`` module.
"""
self.filename = None
if paths:
self.filename = os.path.join(*paths)
if 'root_module' in kwargs:
self.root_module = kwargs['root_module']
else:
self.root_module = awscli
class BasicCommand(CLICommand):
"""Basic top level command with no subcommands.
If you want to create a new command, subclass this and
provide the values documented below.
"""
# This is the name of your command, so if you want to
# create an 'aws mycommand ...' command, the NAME would be
# 'mycommand'
NAME = 'commandname'
# This is the description that will be used for the 'help'
# command.
DESCRIPTION = 'describe the command'
# This is optional, if you are fine with the default synopsis
# (the way all the built in operations are documented) then you
# can leave this empty.
SYNOPSIS = ''
# If you want to provide some hand written examples, you can do
# so here. This is written in RST format. This is optional,
# you don't have to provide any examples, though highly encouraged!
EXAMPLES = ''
# If your command has arguments, you can specify them here. This is
# somewhat of an implementation detail, but this is a list of dicts
# where the dicts match the kwargs of the CustomArgument's __init__.
# For example, if I want to add a '--argument-one' and an
# '--argument-two' command, I'd say:
#
# ARG_TABLE = [
# {'name': 'argument-one', 'help_text': 'This argument does foo bar.',
# 'action': 'store', 'required': False, 'cli_type_name': 'string',},
# {'name': 'argument-two', 'help_text': 'This argument does some other thing.',
# 'action': 'store', 'choices': ['a', 'b', 'c']},
# ]
#
# A `schema` parameter option is available to accept a custom JSON
# structure as input. See the file `awscli/schema.py` for more info.
ARG_TABLE = []
# If you want the command to have subcommands, you can provide a list of
# dicts. We use a list here because we want to allow a user to provide
# the order they want to use for subcommands.
# SUBCOMMANDS = [
# {'name': 'subcommand1', 'command_class': SubcommandClass},
# {'name': 'subcommand2', 'command_class': SubcommandClass2},
# ]
# The command_class must subclass from ``BasicCommand``.
SUBCOMMANDS = []
FROM_FILE = _FromFile
# You can set the DESCRIPTION, SYNOPSIS, and EXAMPLES to FROM_FILE
# and we'll automatically read in that data from the file.
# This is useful if you have a lot of content and would prefer to keep
# the docs out of the class definition. For example:
#
# DESCRIPTION = FROM_FILE
#
# will set the DESCRIPTION value to the contents of
# awscli/examples/<command name>/_description.rst
# The naming conventions for these attributes are:
#
# DESCRIPTION = awscli/examples/<command name>/_description.rst
# SYNOPSIS = awscli/examples/<command name>/_synopsis.rst
# EXAMPLES = awscli/examples/<command name>/_examples.rst
#
# You can also provide a relative path and we'll load the file
# from the specified location:
#
# DESCRIPTION = awscli/examples/<filename>
#
# For example:
#
# DESCRIPTION = FROM_FILE('command, 'subcommand, '_description.rst')
# DESCRIPTION = 'awscli/examples/command/subcommand/_description.rst'
#
# At this point, the only other thing you have to implement is a _run_main
# method (see the method for more information).
def __init__(self, session):
self._session = session
self._arg_table = None
self._subcommand_table = None
self._lineage = [self]
def __call__(self, args, parsed_globals):
# args is the remaining unparsed args.
# We might be able to parse these args so we need to create
# an arg parser and parse them.
self._subcommand_table = self._build_subcommand_table()
self._arg_table = self._build_arg_table()
event = 'before-building-argument-table-parser.%s' % \
".".join(self.lineage_names)
self._session.emit(event, argument_table=self._arg_table, args=args,
session=self._session, parsed_globals=parsed_globals)
parser = ArgTableArgParser(self.arg_table, self.subcommand_table)
parsed_args, remaining = parser.parse_known_args(args)
# Unpack arguments
for key, value in vars(parsed_args).items():
cli_argument = None
# Convert the name to use dashes instead of underscore
# as these are how the parameters are stored in the
# `arg_table`.
xformed = key.replace('_', '-')
if xformed in self.arg_table:
cli_argument = self.arg_table[xformed]
value = unpack_argument(
self._session,
'custom',
self.name,
cli_argument,
value
)
# If this parameter has a schema defined, then allow plugins
# a chance to process and override its value.
if self._should_allow_plugins_override(cli_argument, value):
override = self._session\
.emit_first_non_none_response(
'process-cli-arg.%s.%s' % ('custom', self.name),
cli_argument=cli_argument, value=value, operation=None)
if override is not None:
# A plugin supplied a conversion
value = override
else:
# Unpack the argument, which is a string, into the
# correct Python type (dict, list, etc)
value = unpack_cli_arg(cli_argument, value)
self._validate_value_against_schema(
cli_argument.argument_model, value)
setattr(parsed_args, key, value)
if hasattr(parsed_args, 'help'):
self._display_help(parsed_args, parsed_globals)
elif getattr(parsed_args, 'subcommand', None) is None:
# No subcommand was specified so call the main
# function for this top level command.
if remaining:
raise ValueError("Unknown options: %s" % ','.join(remaining))
return self._run_main(parsed_args, parsed_globals)
else:
return self.subcommand_table[parsed_args.subcommand](remaining,
parsed_globals)
def _validate_value_against_schema(self, model, value):
validate_parameters(value, model)
def _should_allow_plugins_override(self, param, value):
if (param and param.argument_model is not None and
value is not None):
return True
return False
def _run_main(self, parsed_args, parsed_globals):
# Subclasses should implement this method.
# parsed_globals are the parsed global args (things like region,
# profile, output, etc.)
# parsed_args are any arguments you've defined in your ARG_TABLE
# that are parsed. These will come through as whatever you've
# provided as the 'dest' key. Otherwise they default to the
# 'name' key. For example: ARG_TABLE[0] = {"name": "foo-arg", ...}
# can be accessed by ``parsed_args.foo_arg``.
raise NotImplementedError("_run_main")
def _build_subcommand_table(self):
subcommand_table = OrderedDict()
for subcommand in self.SUBCOMMANDS:
subcommand_name = subcommand['name']
subcommand_class = subcommand['command_class']
subcommand_table[subcommand_name] = subcommand_class(self._session)
self._session.emit('building-command-table.%s' % self.NAME,
command_table=subcommand_table,
session=self._session,
command_object=self)
self._add_lineage(subcommand_table)
return subcommand_table
def _display_help(self, parsed_args, parsed_globals):
help_command = self.create_help_command()
help_command(parsed_args, parsed_globals)
def create_help_command(self):
command_help_table = {}
if self.SUBCOMMANDS:
command_help_table = self.create_help_command_table()
return BasicHelp(self._session, self, command_table=command_help_table,
arg_table=self.arg_table)
def create_help_command_table(self):
"""
Create the command table into a form that can be handled by the
BasicDocHandler.
"""
commands = {}
for command in self.SUBCOMMANDS:
commands[command['name']] = command['command_class'](self._session)
self._add_lineage(commands)
return commands
def _build_arg_table(self):
arg_table = OrderedDict()
self._session.emit('building-arg-table.%s' % self.NAME,
arg_table=self.ARG_TABLE)
for arg_data in self.ARG_TABLE:
# If a custom schema was passed in, create the argument_model
# so that it can be validated and docs can be generated.
if 'schema' in arg_data:
argument_model = create_argument_model_from_schema(
arg_data.pop('schema'))
arg_data['argument_model'] = argument_model
custom_argument = CustomArgument(**arg_data)
arg_table[arg_data['name']] = custom_argument
return arg_table
def _add_lineage(self, command_table):
for command in command_table:
command_obj = command_table[command]
command_obj.lineage = self.lineage + [command_obj]
@property
def arg_table(self):
if self._arg_table is None:
self._arg_table = self._build_arg_table()
return self._arg_table
@property
def subcommand_table(self):
if self._subcommand_table is None:
self._subcommand_table = self._build_subcommand_table()
return self._subcommand_table
@classmethod
def add_command(cls, command_table, session, **kwargs):
command_table[cls.NAME] = cls(session)
@property
def name(self):
return self.NAME
@property
def lineage(self):
return self._lineage
@lineage.setter
def lineage(self, value):
self._lineage = value
class BasicHelp(HelpCommand):
def __init__(self, session, command_object, command_table, arg_table,
event_handler_class=None):
super(BasicHelp, self).__init__(session, command_object,
command_table, arg_table)
# This is defined in HelpCommand so we're matching the
# casing here.
if event_handler_class is None:
event_handler_class = BasicDocHandler
self.EventHandlerClass = event_handler_class
# These are public attributes that are mapped from the command
# object. These are used by the BasicDocHandler below.
self._description = command_object.DESCRIPTION
self._synopsis = command_object.SYNOPSIS
self._examples = command_object.EXAMPLES
@property
def name(self):
return self.obj.NAME
@property
def description(self):
return self._get_doc_contents('_description')
@property
def synopsis(self):
return self._get_doc_contents('_synopsis')
@property
def examples(self):
return self._get_doc_contents('_examples')
@property
def event_class(self):
return '.'.join(self.obj.lineage_names)
def _get_doc_contents(self, attr_name):
value = getattr(self, attr_name)
if isinstance(value, BasicCommand.FROM_FILE):
if value.filename is not None:
trailing_path = value.filename
else:
trailing_path = os.path.join(self.name, attr_name + '.rst')
root_module = value.root_module
doc_path = os.path.join(
os.path.abspath(os.path.dirname(root_module.__file__)),
'examples', trailing_path)
with _open(doc_path) as f:
return f.read()
else:
return value
def __call__(self, args, parsed_globals):
# Create an event handler for a Provider Document
instance = self.EventHandlerClass(self)
# Now generate all of the events for a Provider document.
# We pass ourselves along so that we can, in turn, get passed
# to all event handlers.
docevents.generate_events(self.session, self)
self.renderer.render(self.doc.getvalue())
instance.unregister()
class BasicDocHandler(OperationDocumentEventHandler):
def __init__(self, help_command):
super(BasicDocHandler, self).__init__(help_command)
self.doc = help_command.doc
def doc_description(self, help_command, **kwargs):
self.doc.style.h2('Description')
self.doc.write(help_command.description)
self.doc.style.new_paragraph()
def doc_synopsis_start(self, help_command, **kwargs):
if not help_command.synopsis:
super(BasicDocHandler, self).doc_synopsis_start(
help_command=help_command, **kwargs)
else:
self.doc.style.h2('Synopsis')
self.doc.style.start_codeblock()
self.doc.writeln(help_command.synopsis)
def doc_synopsis_option(self, arg_name, help_command, **kwargs):
if not help_command.synopsis:
doc = help_command.doc
argument = help_command.arg_table[arg_name]
if argument.synopsis:
option_str = argument.synopsis
elif argument.group_name in self._arg_groups:
if argument.group_name in self._documented_arg_groups:
# This arg is already documented so we can move on.
return
option_str = ' | '.join(
[a.cli_name for a in
self._arg_groups[argument.group_name]])
self._documented_arg_groups.append(argument.group_name)
elif argument.cli_type_name == 'boolean':
option_str = '%s' % argument.cli_name
elif argument.nargs == '+':
option_str = "%s <value> [<value>...]" % argument.cli_name
else:
option_str = '%s <value>' % argument.cli_name
if not (argument.required or argument.positional_arg):
option_str = '[%s]' % option_str
doc.writeln('%s' % option_str)
else:
# A synopsis has been provided so we don't need to write
# anything here.
pass
def doc_synopsis_end(self, help_command, **kwargs):
if not help_command.synopsis and not help_command.command_table:
super(BasicDocHandler, self).doc_synopsis_end(
help_command=help_command, **kwargs)
else:
self.doc.style.end_codeblock()
def doc_global_option(self, help_command, **kwargs):
if not help_command.command_table:
super().doc_global_option(help_command, **kwargs)
def doc_examples(self, help_command, **kwargs):
if help_command.examples:
self.doc.style.h2('Examples')
self.doc.write(help_command.examples)
def doc_subitems_start(self, help_command, **kwargs):
if help_command.command_table:
doc = help_command.doc
doc.style.h2('Available Commands')
doc.style.toctree()
def doc_subitem(self, command_name, help_command, **kwargs):
if help_command.command_table:
doc = help_command.doc
doc.style.tocitem(command_name)
def doc_subitems_end(self, help_command, **kwargs):
pass
def doc_output(self, help_command, event_name, **kwargs):
pass