python类xform_name()的实例源码

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 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
        )

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 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)