Class CLIOperationCaller

awscli/clidriver.py:1056–1117

Call an AWS operation and format the response.

class CLIOperationCaller:
    """Call an AWS operation and format the response."""

    def __init__(self, session):
        self._session = session
        self._output_stream_factory = OutputStreamFactory(session)

    def invoke(self, service_name, operation_name, parameters, parsed_globals):
        """Invoke an operation and format the response.

        :type service_name: str
        :param service_name: The name of the service.  Note this is the
            service name, not the endpoint prefix (e.g. ``ses`` not ``email``).

        :type operation_name: str
        :param operation_name: The operation name of the service.  The casing
            of the operation name should match the exact casing used by the
            service, e.g. ``DescribeInstances``, not ``describe-instances`` or
            ``describe_instances``.

        :type parameters: dict
        :param parameters: The parameters for the operation call.  Again,
            these values have the same casing used by the service.

        :type parsed_globals: Namespace
        :param parsed_globals: The parsed globals from the command line.

        :return: None, the result is displayed through a formatter, but no
            value is returned.

        """
        client = self._session.create_client(
            service_name,
            region_name=parsed_globals.region,
            endpoint_url=parsed_globals.endpoint_url,
            verify=parsed_globals.verify_ssl,
        )
        response = self._make_client_call(
            client, operation_name, parameters, parsed_globals
        )
        self._display_response(operation_name, response, parsed_globals)
        return 0

    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, py_operation_name)(**parameters)
        return response

    def _display_response(self, command_name, response, parsed_globals):
        output = parsed_globals.output
        if output is None:
            output = self._session.get_config_variable('output')