我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用botocore.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 __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)
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)
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 _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 )
