def __call__(self, parent, *args, **kwargs):
"""
Perform the wait operation after building operation
parameters.
:type parent: :py:class:`~boto3.resources.base.ServiceResource`
:param parent: The resource instance to which this action is attached.
"""
client_waiter_name = xform_name(self._waiter_model.waiter_name)
# 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._waiter_model)
params.update(kwargs)
logger.info('Calling %s:%s with %r',
parent.meta.service_name,
self._waiter_resource_name, params)
client = parent.meta.client
waiter = client.get_waiter(client_waiter_name)
response = waiter.wait(**params)
logger.debug('Response: %r', response)
python类xform_name()的实例源码
def _add_single_waiter(self, section, waiter_name):
section = section.add_new_section(waiter_name)
section.style.start_sphinx_py_class(
class_name='%s.Waiter.%s' % (
self._client.__class__.__name__, waiter_name))
# Add example on how to instantiate waiter.
section.style.start_codeblock()
section.style.new_line()
section.write(
'waiter = client.get_waiter(\'%s\')' % xform_name(waiter_name)
)
section.style.end_codeblock()
# Add information on the wait() method
section.style.new_line()
document_wait_method(
section=section,
waiter_name=waiter_name,
event_emitter=self._client.meta.events,
service_model=self._client.meta.service_model,
service_waiter_model=self._service_waiter_model
)
def build_translation_map(self):
operation_model = self.help_command.obj
d = {}
for cli_name, cli_argument in self.help_command.arg_table.items():
if cli_argument.argument_model is not None:
argument_name = cli_argument.argument_model.name
if argument_name in d:
previous_mapping = d[argument_name]
# If the argument name is a boolean argument, we want the
# the translation to default to the one that does not start
# with --no-. So we check if the cli parameter currently
# being used starts with no- and if stripping off the no-
# results in the new proposed cli argument name. If it
# does, we assume we have the postive form of the argument
# which is the name we want to use in doc translations.
if cli_argument.cli_type_name == 'boolean' and \
previous_mapping.startswith('no-') and \
cli_name == previous_mapping[3:]:
d[argument_name] = cli_name
else:
d[argument_name] = cli_name
for operation_name in operation_model.service_model.operation_names:
d[operation_name] = xform_name(operation_name, '-')
return d
def _create_command_table(self):
command_table = OrderedDict()
service_model = self._get_service_model()
for operation_name in service_model.operation_names:
cli_name = xform_name(operation_name, '-')
operation_model = service_model.operation_model(operation_name)
command_table[cli_name] = ServiceOperation(
name=cli_name,
parent_name=self._name,
session=self.session,
operation_model=operation_model,
operation_caller=CLIOperationCaller(self.session),
)
self.session.emit('building-command-table.%s' % self._name,
command_table=command_table,
session=self.session,
command_object=self)
self._add_lineage(command_table)
return command_table
def _add_single_waiter(self, section, waiter_name):
section = section.add_new_section(waiter_name)
section.style.start_sphinx_py_class(
class_name='%s.Waiter.%s' % (
self._client.__class__.__name__, waiter_name))
# Add example on how to instantiate waiter.
section.style.start_codeblock()
section.style.new_line()
section.write(
'waiter = client.get_waiter(\'%s\')' % xform_name(waiter_name)
)
section.style.end_codeblock()
# Add information on the wait() method
section.style.new_line()
document_wait_method(
section=section,
waiter_name=waiter_name,
event_emitter=self._client.meta.events,
service_model=self._client.meta.service_model,
service_waiter_model=self._service_waiter_model
)
def __call__(self, parent, *args, **kwargs):
"""
Perform the wait operation after building operation
parameters.
:type parent: :py:class:`~boto3.resources.base.ServiceResource`
:param parent: The resource instance to which this action is attached.
"""
client_waiter_name = xform_name(self._waiter_model.waiter_name)
# 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._waiter_model)
params.update(kwargs)
logger.info('Calling %s:%s with %r',
parent.meta.service_name,
self._waiter_resource_name, params)
client = parent.meta.client
waiter = client.get_waiter(client_waiter_name)
response = waiter.wait(**params)
logger.debug('Response: %r', response)
def _add_single_waiter(self, section, waiter_name):
section = section.add_new_section(waiter_name)
section.style.start_sphinx_py_class(
class_name='%s.Waiter.%s' % (
self._client.__class__.__name__, waiter_name))
# Add example on how to instantiate waiter.
section.style.start_codeblock()
section.style.new_line()
section.write(
'waiter = client.get_waiter(\'%s\')' % xform_name(waiter_name)
)
section.style.end_codeblock()
# Add information on the wait() method
section.style.new_line()
document_wait_method(
section=section,
waiter_name=waiter_name,
event_emitter=self._client.meta.events,
service_model=self._client.meta.service_model,
service_waiter_model=self._service_waiter_model
)
def __call__(self, parent, *args, **kwargs):
"""
Perform the wait operation after building operation
parameters.
:type parent: :py:class:`~boto3.resources.base.ServiceResource`
:param parent: The resource instance to which this action is attached.
"""
client_waiter_name = xform_name(self._waiter_model.waiter_name)
# 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._waiter_model)
params.update(kwargs)
logger.info('Calling %s:%s with %r',
parent.meta.service_name,
self._waiter_resource_name, params)
client = parent.meta.client
waiter = client.get_waiter(client_waiter_name)
response = waiter.wait(**params)
logger.debug('Response: %r', response)
def _add_single_waiter(self, section, waiter_name):
section = section.add_new_section(waiter_name)
section.style.start_sphinx_py_class(
class_name='%s.Waiter.%s' % (
self._client.__class__.__name__, waiter_name))
# Add example on how to instantiate waiter.
section.style.start_codeblock()
section.style.new_line()
section.write(
'waiter = client.get_waiter(\'%s\')' % xform_name(waiter_name)
)
section.style.end_codeblock()
# Add information on the wait() method
section.style.new_line()
document_wait_method(
section=section,
waiter_name=waiter_name,
event_emitter=self._client.meta.events,
service_model=self._client.meta.service_model,
service_waiter_model=self._service_waiter_model
)
def __call__(self, parent, *args, **kwargs):
"""
Perform the wait operation after building operation
parameters.
:type parent: :py:class:`~boto3.resources.base.ServiceResource`
:param parent: The resource instance to which this action is attached.
"""
client_waiter_name = xform_name(self._waiter_model.waiter_name)
# 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._waiter_model)
params.update(kwargs)
logger.info('Calling %s:%s with %r',
parent.meta.service_name,
self._waiter_resource_name, params)
client = parent.meta.client
waiter = client.get_waiter(client_waiter_name)
response = waiter.wait(**params)
logger.debug('Response: %r', response)
def _add_single_waiter(self, section, waiter_name):
section = section.add_new_section(waiter_name)
section.style.start_sphinx_py_class(
class_name='%s.Waiter.%s' % (
self._client.__class__.__name__, waiter_name))
# Add example on how to instantiate waiter.
section.style.start_codeblock()
section.style.new_line()
section.write(
'waiter = client.get_waiter(\'%s\')' % xform_name(waiter_name)
)
section.style.end_codeblock()
# Add information on the wait() method
section.style.new_line()
document_wait_method(
section=section,
waiter_name=waiter_name,
event_emitter=self._client.meta.events,
service_model=self._client.meta.service_model,
service_waiter_model=self._service_waiter_model
)
action.py 文件源码
项目:tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 34
收藏 0
点赞 0
评论 0
def __call__(self, parent, *args, **kwargs):
"""
Perform the wait operation after building operation
parameters.
:type parent: :py:class:`~boto3.resources.base.ServiceResource`
:param parent: The resource instance to which this action is attached.
"""
client_waiter_name = xform_name(self._waiter_model.waiter_name)
# 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._waiter_model)
params.update(kwargs)
logger.info('Calling %s:%s with %r',
parent.meta.service_name,
self._waiter_resource_name, params)
client = parent.meta.client
waiter = client.get_waiter(client_waiter_name)
response = waiter.wait(**params)
logger.debug('Response: %r', response)
waiter.py 文件源码
项目:tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 24
收藏 0
点赞 0
评论 0
def _add_single_waiter(self, section, waiter_name):
section = section.add_new_section(waiter_name)
section.style.start_sphinx_py_class(
class_name='%s.Waiter.%s' % (
self._client.__class__.__name__, waiter_name))
# Add example on how to instantiate waiter.
section.style.start_codeblock()
section.style.new_line()
section.write(
'waiter = client.get_waiter(\'%s\')' % xform_name(waiter_name)
)
section.style.end_codeblock()
# Add information on the wait() method
section.style.new_line()
document_wait_method(
section=section,
waiter_name=waiter_name,
event_emitter=self._client.meta.events,
service_model=self._client.meta.service_model,
service_waiter_model=self._service_waiter_model
)
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
)
def _get_all_cli_input_tokens(pagination_config):
# Get all input tokens including the limit_key
# if it exists.
tokens = _get_input_tokens(pagination_config)
for token_name in tokens:
cli_name = xform_name(token_name, '-')
yield cli_name
if 'limit_key' in pagination_config:
key_name = pagination_config['limit_key']
cli_name = xform_name(key_name, '-')
yield cli_name
def _make_client_call(self, client, operation_name, parameters,
parsed_globals):
method_name = xform_name(operation_name)
operation_model = client.meta.service_model.operation_model(
operation_name)
fake_response = {}
if operation_model.output_shape:
argument_generator = ArgumentGenerator(use_member_names=True)
fake_response = argument_generator.generate_skeleton(
operation_model.output_shape)
with Stubber(client) as stubber:
stubber.add_response(method_name, fake_response)
return getattr(client, method_name)(**parameters)
def build_all_waiter_state_cmds(self, subcommand_table):
"""This adds waiter state commands to the subcommand table passed in.
This is the method that adds waiter state commands like
``instance-running`` to ``ec2 wait``.
"""
waiter_names = self._model.waiter_names
for waiter_name in waiter_names:
waiter_cli_name = xform_name(waiter_name, '-')
subcommand_table[waiter_cli_name] = \
self._build_waiter_state_cmd(waiter_name)
def _build_waiter_state_cmd(self, waiter_name):
# Get the waiter
waiter_config = self._model.get_waiter(waiter_name)
# Create the cli name for the waiter operation
waiter_cli_name = xform_name(waiter_name, '-')
# Obtain the name of the service operation that is used to implement
# the specified waiter.
operation_name = waiter_config.operation
# Create an operation object to make a command for the waiter. The
# operation object is used to generate the arguments for the waiter
# state command.
operation_model = self._service_model.operation_model(operation_name)
waiter_state_command = WaiterStateCommand(
name=waiter_cli_name, parent_name='wait',
operation_caller=WaiterCaller(self._session, waiter_name),
session=self._session,
operation_model=operation_model,
)
# Build the top level description for the waiter state command.
# Most waiters do not have a description so they need to be generated
# using the waiter configuration.
waiter_state_doc_builder = WaiterStateDocBuilder(waiter_config)
description = waiter_state_doc_builder.build_waiter_state_description()
waiter_state_command.DESCRIPTION = description
return waiter_state_command
def _build_operation_description(self, operation):
operation_name = xform_name(operation).replace('_', '-')
return u'when polling with ``%s``.' % operation_name
def _call_iam_operation(self, operation_name, parameters, parsed_globals):
client = self._session.create_client(
'iam', region_name=self.region, endpoint_url=self.iam_endpoint_url,
verify=parsed_globals.verify_ssl)
return getattr(client, xform_name(operation_name))(**parameters)