def __init__(self, cli_name, message):

"""

full_message = ("Error parsing parameter '%s': %s" %

(cli_name, message))

super(ParamError, self).__init__(full_message)

self.cli_name = cli_name

def __init__(self, key, valid_keys):

valid_keys = ', '.join(valid_keys)

full_message = (

"Unknown key '%s', valid choices "

"are: %s" % (key, valid_keys))

"""

param_name = getattr(cli_argument, 'name', 'anonymous')

value_override = session.emit_first_non_none_response(

'load-cli-arg.%s.%s.%s' % (service_name,

operation_name,

param_name),

param=cli_argument, value=value, service_name=service_name,

operation_name=operation_name)

if value_override is not None:

value = value_override

"""Handler that supports param values from URIs.

cli_argument = param

qualified_param_name = '.'.join(event_name.split('.')[1:])

if qualified_param_name in PARAMFILE_DISABLED or \

getattr(cli_argument, 'no_paramfile', None):

return

else:

if isinstance(value, list) and len(value) == 1:

value = value[0]

try:

return get_paramfile(value)

except ResourceLoadingError as e:

if param.name in stack:

return 'recursive'

else:

stack.append(param.name)

try:

if param.type_name in SCALAR_TYPES:

return 'scalar'

elif param.type_name == 'structure':

sub_types = [_detect_shape_structure(p, stack)

for p in param.members.values()]

# We're distinguishing between structure(scalar)

# and structure(scalars), because for the case of

# a single scalar in a structure we can simplify

# more than a structure(scalars).

if len(sub_types) == 1 and all(p == 'scalar' for p in sub_types):

return 'structure(scalar)'

elif len(sub_types) > 1 and all(p == 'scalar' for p in sub_types):

return 'structure(scalars)'

else:

return 'structure(%s)' % ', '.join(sorted(set(sub_types)))

elif param.type_name == 'list':

return 'list-%s' % _detect_shape_structure(param.member, stack)

elif param.type_name == 'map':

if param.value.type_name in SCALAR_TYPES:

return 'map-scalar'

else:

return 'map-%s' % _detect_shape_structure(param.value, stack)

finally:

"""

return _unpack_cli_arg(cli_argument.argument_model, value,

# check if model is jsonvalue header and that value is serializable

if model.serialization.get('jsonvalue') and \

model.serialization.get('location') == 'header' and \

model.type_name == 'string':

return True

if is_json_value_header(argument_model):

return _unpack_json_cli_arg(argument_model, value, cli_name)

elif argument_model.type_name in SCALAR_TYPES:

return unpack_scalar_cli_arg(

argument_model, value, cli_name)

elif argument_model.type_name in COMPLEX_TYPES:

return _unpack_complex_cli_arg(

argument_model, value, cli_name)

else:

try:

return json.loads(value, object_pairs_hook=OrderedDict)

except ValueError as e:

raise ParamError(

cli_name, "Invalid JSON: %s\nJSON received: %s"

type_name = argument_model.type_name

if type_name == 'structure' or type_name == 'map':

if value.lstrip()[0] == '{':

try:

return json.loads(value, object_pairs_hook=OrderedDict)

except ValueError as e:

raise ParamError(

cli_name, "Invalid JSON: %s\nJSON received: %s"

% (e, value))

raise ParamError(cli_name, "Invalid JSON:\n%s" % value)

elif type_name == 'list':

if isinstance(value, six.string_types):

if value.lstrip()[0] == '[':

return json.loads(value, object_pairs_hook=OrderedDict)

elif isinstance(value, list) and len(value) == 1:

single_value = value[0].strip()

if single_value and single_value[0] == '[':

return json.loads(value[0], object_pairs_hook=OrderedDict)

try:

# There's a couple of cases remaining here.

# 1. It's possible that this is just a list of strings, i.e

# --security-group-ids sg-1 sg-2 sg-3 => ['sg-1', 'sg-2', 'sg-3']

# 2. It's possible this is a list of json objects:

# --filters '{"Name": ..}' '{"Name": ...}'

member_shape_model = argument_model.member

return [_unpack_cli_arg(member_shape_model, v, cli_name)

for v in value]

except (ValueError, TypeError) as e:

# The list params don't have a name/cli_name attached to them

# so they will have bad error messages. We're going to

# attach the parent parameter to this error message to provide

# a more helpful error message.

# Note the cli_name is used strictly for error reporting. It's

# not required to use unpack_scalar_cli_arg

if argument_model.type_name == 'integer' or argument_model.type_name == 'long':

return int(value)

elif argument_model.type_name == 'float' or argument_model.type_name == 'double':

# TODO: losing precision on double types

return float(value)

elif argument_model.type_name == 'blob' and \

argument_model.serialization.get('streaming'):

file_path = os.path.expandvars(value)

file_path = os.path.expanduser(file_path)

if not os.path.isfile(file_path):

msg = 'Blob values must be a path to a file.'

raise ParamError(cli_name, msg)

return open(file_path, 'rb')

elif argument_model.type_name == 'boolean':

if isinstance(value, six.string_types) and value.lower() == 'false':

return False

return bool(value)

else:

if model.type_name not in ['structure', 'list', 'map']:

return False

elif model.type_name == 'list':

if model.member.type_name not in ['structure', 'list', 'map']:

return False

def _uses_old_list_case(self, service_name, operation_name, argument_name):

"""

cases = {

'firehose': {

'put-record-batch': ['records']

},

'workspaces': {

'reboot-workspaces': ['reboot-workspace-requests'],

'rebuild-workspaces': ['rebuild-workspace-requests'],

'terminate-workspaces': ['terminate-workspace-requests']

},

'elb': {

'remove-tags': ['tags'],

'describe-instance-health': ['instances'],

'deregister-instances-from-load-balancer': ['instances'],

'register-instances-with-load-balancer': ['instances']

}

}

cases = cases.get(service_name, {}).get(operation_name, [])

def __init__(self):

self._parser = shorthand.ShorthandParser()

self._visitor = shorthand.BackCompatVisitor()

def __call__(self, cli_argument, value, event_name, **kwargs):

"""Attempt to parse shorthand syntax for values.

if not self._should_parse_as_shorthand(cli_argument, value):

return

else:

service_name, operation_name = \

find_service_and_method_in_event_name(event_name)

return self._parse_as_shorthand(

cli_argument, value, service_name, operation_name)

def _parse_as_shorthand(self, cli_argument, value, service_name,

operation_name):

try:

LOG.debug("Parsing param %s as shorthand",

cli_argument.cli_name)

handled_value = self._handle_special_cases(

cli_argument, value, service_name, operation_name)

if handled_value is not None:

return handled_value

if isinstance(value, list):

# Because of how we're using argparse, list shapes

# are configured with nargs='+' which means the ``value``

# is given to us "conveniently" as a list. When

# this happens we need to parse each list element

# individually.

parsed = [self._parser.parse(v) for v in value]

self._visitor.visit(parsed, cli_argument.argument_model)

else:

# Otherwise value is just a string.

parsed = self._parser.parse(value)

self._visitor.visit(parsed, cli_argument.argument_model)

except shorthand.ShorthandParseError as e:

raise ParamError(cli_argument.cli_name, str(e))

except (ParamError, ParamUnknownKeyError) as e:

# The shorthand parse methods don't have the cli_name,

# so any ParamError won't have this value. To accomodate

# this, ParamErrors are caught and reraised with the cli_name

# injected.

raise ParamError(cli_argument.cli_name, str(e))

return parsed

def _handle_special_cases(self, cli_argument, value, service_name,

operation_name):

# We need to handle a few special cases that the previous

# parser handled in order to stay backwards compatible.

model = cli_argument.argument_model

if model.type_name == 'list' and \

model.member.type_name == 'structure' and \

len(model.member.members) == 1 and \

self._uses_old_list_case(service_name, operation_name, cli_argument.name):

# First special case is handling a list of structures

# of a single element such as:

#

# --instance-ids id-1 id-2 id-3

#

# gets parsed as:

#

# [{"InstanceId": "id-1"}, {"InstanceId": "id-2"},

# {"InstanceId": "id-3"}]

key_name = list(model.member.members.keys())[0]

new_values = [{key_name: v} for v in value]

return new_values

elif model.type_name == 'structure' and \

len(model.members) == 1 and \

'Value' in model.members and \

model.members['Value'].type_name == 'string' and \

'=' not in value:

# Second special case is where a structure of a single

# value whose member name is "Value" can be specified

# as:

# --instance-terminate-behavior shutdown

#

# gets parsed as:

# {"Value": "shutdown"}

return {'Value': value}

def _should_parse_as_shorthand(self, cli_argument, value):

# We first need to make sure this is a parameter that qualifies

# for simplification. The first short-circuit case is if it looks

# like json we immediately return.

if value and isinstance(value, list):

check_val = value[0]

else:

check_val = value

if isinstance(check_val, six.string_types) and check_val.strip().startswith(

('[', '{')):

LOG.debug("Param %s looks like JSON, not considered for "

"param shorthand.", cli_argument.py_name)

return False

model = cli_argument.argument_model

# The second case is to make sure the argument is sufficiently

# complex, that is, it's base type is a complex type *and*

# if it's a list, then it can't be a list of scalar types.

"""Documentation generator for param shorthand syntax."""

_DONT_DOC = object()

_MAX_STACK = 3

def supports_shorthand(self, argument_model):

"""Checks if a CLI argument supports shorthand syntax."""

if argument_model is not None:

return _is_complex_shape(argument_model)

return False

def generate_shorthand_example(self, cli_argument, service_name,

operation_name):

"""Generate documentation for a CLI argument.

docstring = self._handle_special_cases(

cli_argument, service_name, operation_name)

if docstring is self._DONT_DOC:

return None

elif docstring:

return docstring

# Otherwise we fall back to the normal docgen for shorthand

# syntax.

stack = []

try:

if cli_argument.argument_model.type_name == 'list':

argument_model = cli_argument.argument_model.member

return self._shorthand_docs(argument_model, stack) + ' ...'

else:

return self._shorthand_docs(cli_argument.argument_model, stack)

except TooComplexError:

return ''

def _handle_special_cases(self, cli_argument, service_name, operation_name):

model = cli_argument.argument_model

if model.type_name == 'list' and \

model.member.type_name == 'structure' and \

len(model.member.members) == 1 and \

self._uses_old_list_case(

service_name, operation_name, cli_argument.name):

member_name = list(model.member.members)[0]

return '%s %s1 %s2 %s3' % (cli_argument.cli_name, member_name,

member_name, member_name)

elif model.type_name == 'structure' and \

len(model.members) == 1 and \

'Value' in model.members and \

model.members['Value'].type_name == 'string':

return self._DONT_DOC

return ''

def _shorthand_docs(self, argument_model, stack):

if len(stack) > self._MAX_STACK:

raise TooComplexError()

if argument_model.type_name == 'structure':

return self._structure_docs(argument_model, stack)

elif argument_model.type_name == 'list':

return self._list_docs(argument_model, stack)

elif argument_model.type_name == 'map':

return self._map_docs(argument_model, stack)

else:

return argument_model.type_name

def _list_docs(self, argument_model, stack):

list_member = argument_model.member

stack.append(list_member.name)

try:

element_docs = self._shorthand_docs(argument_model.member, stack)

finally:

stack.pop()

if list_member.type_name in COMPLEX_TYPES or len(stack) > 1:

return '[%s,%s]' % (element_docs, element_docs)

else:

return '%s,%s' % (element_docs, element_docs)

def _map_docs(self, argument_model, stack):

k = argument_model.key

value_docs = self._shorthand_docs(argument_model.value, stack)

start = 'KeyName1=%s,KeyName2=%s' % (value_docs, value_docs)

if k.enum and not stack:

start += '\n\nWhere valid key names are:\n'

for enum in k.enum:

start += ' %s\n' % enum

elif stack:

start = '{%s}' % start

return start

def _structure_docs(self, argument_model, stack):

parts = []

for name, member_shape in argument_model.members.items():

parts.append(self._member_docs(name, member_shape, stack))

inner_part = ','.join(parts)

if not stack:

return inner_part

return '{%s}' % inner_part

def _member_docs(self, name, shape, stack):

if stack.count(shape.name) > 0:

return '( ... recursive ... )'

stack.append(shape.name)

try:

value_doc = self._shorthand_docs(shape, stack)

finally:

stack.pop()