def _unpack_argument(self, value):
service_name = self._operation_model.service_model.service_name
operation_name = xform_name(self._operation_model.name, '-')
override = self._emit_first_response('process-cli-arg.%s.%s' % (
service_name, operation_name), param=self.argument_model,
cli_argument=self, value=value)
if override is not None:
# A plugin supplied an alternate conversion,
# use it instead.
return override
else:
# Fall back to the default arg processing.
return unpack_cli_arg(self, value)
python类xform_name()的实例源码
def build_translation_map(self):
d = {}
service_model = self.help_command.obj
for operation_name in service_model.operation_names:
d[operation_name] = xform_name(operation_name, '-')
return d
# A service document has no synopsis.
def _create_argument_table(self):
argument_table = OrderedDict()
input_shape = self._operation_model.input_shape
required_arguments = []
arg_dict = {}
if input_shape is not None:
required_arguments = input_shape.required_members
arg_dict = input_shape.members
for arg_name, arg_shape in arg_dict.items():
cli_arg_name = xform_name(arg_name, '-')
arg_class = self.ARG_TYPES.get(arg_shape.type_name,
self.DEFAULT_ARG_CLASS)
is_required = arg_name in required_arguments
event_emitter = self._session.get_component('event_emitter')
arg_object = arg_class(
name=cli_arg_name,
argument_model=arg_shape,
is_required=is_required,
operation_model=self._operation_model,
serialized_name=arg_name,
event_emitter=event_emitter)
arg_object.add_to_arg_table(argument_table)
LOG.debug(argument_table)
self._emit('building-argument-table.%s.%s' % (self._parent_name,
self._name),
operation_model=self._operation_model,
session=self._session,
command=self,
argument_table=argument_table)
return argument_table
def _make_client_call(self, client, operation_name, parameters,
parsed_globals):
py_operation_name = xform_name(operation_name)
if client.can_paginate(py_operation_name) and parsed_globals.paginate:
paginator = client.get_paginator(py_operation_name)
response = paginator.paginate(**parameters)
else:
response = getattr(client, xform_name(operation_name))(
**parameters)
return response
def _create_methods(self, service_model):
op_dict = {}
for operation_name in service_model.operation_names:
py_operation_name = xform_name(operation_name)
op_dict[py_operation_name] = self._create_api_method(
py_operation_name, operation_name, service_model)
return op_dict
def _create_name_mapping(self, service_model):
# py_name -> OperationName, for every operation available
# for a service.
mapping = {}
for operation_name in service_model.operation_names:
py_operation_name = xform_name(operation_name)
mapping[py_operation_name] = operation_name
return mapping
def get_waiter(self, waiter_name):
config = self._get_waiter_config()
if not config:
raise ValueError("Waiter does not exist: %s" % waiter_name)
model = waiter.WaiterModel(config)
mapping = {}
for name in model.waiter_names:
mapping[xform_name(name)] = name
if waiter_name not in mapping:
raise ValueError("Waiter does not exist: %s" % waiter_name)
return waiter.create_waiter_with_client(
mapping[waiter_name], model, self)
def _add_paginator(self, section, paginator_name):
section = section.add_new_section(paginator_name)
# Docment the paginator class
section.style.start_sphinx_py_class(
class_name='%s.Paginator.%s' % (
self._client.__class__.__name__, paginator_name))
section.style.start_codeblock()
section.style.new_line()
# Document how to instantiate the paginator.
section.write(
'paginator = client.get_paginator(\'%s\')' % xform_name(
paginator_name)
)
section.style.end_codeblock()
section.style.new_line()
# Get the pagination model for the particular paginator.
paginator_config = self._service_paginator_model.get_paginator(
paginator_name)
document_paginate_method(
section=section,
paginator_name=paginator_name,
event_emitter=self._client.meta.events,
service_model=self._client.meta.service_model,
paginator_config=paginator_config
)
def document_wait_method(section, waiter_name, event_emitter,
service_model, service_waiter_model,
include_signature=True):
"""Documents a the wait method of a waiter
:param section: The section to write to
:param waiter_name: The name of the waiter
:param event_emitter: The event emitter to use to emit events
:param service_model: The service model
:param service_waiter_model: The waiter model associated to the service
:param include_signature: Whether or not to include the signature.
It is useful for generating docstrings.
"""
waiter_model = service_waiter_model.get_waiter(waiter_name)
operation_model = service_model.operation_model(
waiter_model.operation)
wait_description = (
'Polls :py:meth:`{0}.Client.{1}` every {2} '
'seconds until a successful state is reached. An error is '
'returned after {3} failed checks.'.format(
get_service_module_name(service_model),
xform_name(waiter_model.operation),
waiter_model.delay, waiter_model.max_attempts)
)
document_model_driven_method(
section, 'wait', operation_model,
event_emitter=event_emitter,
method_description=wait_description,
example_prefix='waiter.wait',
document_output=False,
include_signature=include_signature
)
def __call__(self, parent, *args, **kwargs):
"""
Perform the action's request operation after building operation
parameters and build any defined resources from the response.
:type parent: :py:class:`~boto3.resources.base.ServiceResource`
:param parent: The resource instance to which this action is attached.
:rtype: dict or ServiceResource or list(ServiceResource)
:return: The response, either as a raw dict or resource instance(s).
"""
operation_name = xform_name(self._action_model.request.operation)
# First, build predefined params and then update with the
# user-supplied kwargs, which allows overriding the pre-built
# params if needed.
params = create_request_parameters(parent, self._action_model.request)
params.update(kwargs)
logger.info('Calling %s:%s with %r', parent.meta.service_name,
operation_name, params)
response = getattr(parent.meta.client, operation_name)(**params)
logger.debug('Response: %r', response)
return self._response_handler(parent, params, response)
def _load_name_with_category(self, names, name, category,
snake_case=True):
"""
Load a name with a given category, possibly renaming it
if that name is already in use. The name will be stored
in ``names`` and possibly be set up in ``self._renamed``.
:type names: set
:param names: Existing names (Python attributes, properties, or
methods) on the resource.
:type name: string
:param name: The original name of the value.
:type category: string
:param category: The value type, such as 'identifier' or 'action'
:type snake_case: bool
:param snake_case: True (default) if the name should be snake cased.
"""
if snake_case:
name = xform_name(name)
if name in names:
logger.debug('Renaming %s %s %s' % (self.name, category, name))
self._renamed[(category, name)] = name + '_' + category
name += '_' + category
if name in names:
# This isn't good, let's raise instead of trying to keep
# renaming this value.
raise ValueError('Problem renaming {0} {1} to {2}!'.format(
self.name, category, name))
names.add(name)
def get_attributes(self, shape):
"""
Get a dictionary of attribute names to original name and shape
models that represent the attributes of this resource. Looks
like the following:
{
'some_name': ('SomeName', <Shape...>)
}
:type shape: botocore.model.Shape
:param shape: The underlying shape for this resource.
:rtype: dict
:return: Mapping of resource attributes.
"""
attributes = {}
identifier_names = [i.name for i in self.identifiers]
for name, member in shape.members.items():
snake_cased = xform_name(name)
if snake_cased in identifier_names:
# Skip identifiers, these are set through other means
continue
snake_cased = self._get_name('attribute', snake_cased,
snake_case=False)
attributes[snake_cased] = (name, member)
return attributes
def __init__(self, model, parent, handler, **kwargs):
self._model = model
self._parent = parent
self._py_operation_name = xform_name(
model.request.operation)
self._handler = handler
self._params = kwargs
def _load_batch_actions(self, attrs, resource_name, collection_model,
service_model, event_emitter):
"""
Batch actions on the collection become methods on both
the collection manager and iterators.
"""
for action_model in collection_model.batch_actions:
snake_cased = xform_name(action_model.name)
attrs[snake_cased] = self._create_batch_action(
resource_name, snake_cased, action_model, collection_model,
service_model, event_emitter)
def document_action(section, resource_name, event_emitter, action_model,
service_model, include_signature=True):
"""Documents a resource action
:param section: The section to write to
:param resource_name: The name of the resource
:param event_emitter: The event emitter to use to emit events
:param action_model: The model of the action
:param service_model: The model of the service
:param include_signature: Whether or not to include the signature.
It is useful for generating docstrings.
"""
operation_model = service_model.operation_model(
action_model.request.operation)
ignore_params = get_resource_ignore_params(action_model.request.params)
example_return_value = 'response'
if action_model.resource:
example_return_value = xform_name(action_model.resource.type)
example_resource_name = xform_name(resource_name)
if service_model.service_name == resource_name:
example_resource_name = resource_name
example_prefix = '%s = %s.%s' % (
example_return_value, example_resource_name, action_model.name)
document_model_driven_resource_method(
section=section, method_name=action_model.name,
operation_model=operation_model,
event_emitter=event_emitter,
method_description=operation_model.documentation,
example_prefix=example_prefix,
exclude_input=ignore_params,
resource_action_model=action_model,
include_signature=include_signature
)
def _create_methods(self, service_model):
op_dict = {}
for operation_name in service_model.operation_names:
py_operation_name = xform_name(operation_name)
op_dict[py_operation_name] = self._create_api_method(
py_operation_name, operation_name, service_model)
return op_dict
def _create_name_mapping(self, service_model):
# py_name -> OperationName, for every operation available
# for a service.
mapping = {}
for operation_name in service_model.operation_names:
py_operation_name = xform_name(operation_name)
mapping[py_operation_name] = operation_name
return mapping
def get_waiter(self, waiter_name):
config = self._get_waiter_config()
if not config:
raise ValueError("Waiter does not exist: %s" % waiter_name)
model = waiter.WaiterModel(config)
mapping = {}
for name in model.waiter_names:
mapping[xform_name(name)] = name
if waiter_name not in mapping:
raise ValueError("Waiter does not exist: %s" % waiter_name)
return waiter.create_waiter_with_client(
mapping[waiter_name], model, self)
def waiter_names(self):
"""Returns a list of all available waiters."""
config = self._get_waiter_config()
if not config:
return []
model = waiter.WaiterModel(config)
# Waiter configs is a dict, we just want the waiter names
# which are the keys in the dict.
return [xform_name(name) for name in model.waiter_names]
def document_wait_method(section, waiter_name, event_emitter,
service_model, service_waiter_model,
include_signature=True):
"""Documents a the wait method of a waiter
:param section: The section to write to
:param waiter_name: The name of the waiter
:param event_emitter: The event emitter to use to emit events
:param service_model: The service model
:param service_waiter_model: The waiter model associated to the service
:param include_signature: Whether or not to include the signature.
It is useful for generating docstrings.
"""
waiter_model = service_waiter_model.get_waiter(waiter_name)
operation_model = service_model.operation_model(
waiter_model.operation)
wait_description = (
'Polls :py:meth:`{0}.Client.{1}` every {2} '
'seconds until a successful state is reached. An error is '
'returned after {3} failed checks.'.format(
get_service_module_name(service_model),
xform_name(waiter_model.operation),
waiter_model.delay, waiter_model.max_attempts)
)
document_model_driven_method(
section, 'wait', operation_model,
event_emitter=event_emitter,
method_description=wait_description,
example_prefix='waiter.wait',
document_output=False,
include_signature=include_signature
)
