Method invoke

zappa/cli.py:1551–1597

Invoke a remote function.

(
        self, function_name, raw_python=False, command=None, no_color=False, client_context=None, qualifier=None, payload=None
    )

            click.echo("SNS Topic removed: %s" % ", ".join(removed_arns))

    def invoke(
        self, function_name, raw_python=False, command=None, no_color=False, client_context=None, qualifier=None, payload=None
    ):
        """
        Invoke a remote function.
        """

        # There are three likely scenarios for 'command' here:
        #   command, which is a modular function path
        #   raw_command, which is a string of python to execute directly
        #   manage, which is a Django-specific management command invocation
        key = command if command is not None else "command"
        if raw_python:
            command = {"raw_command": function_name}
        else:
            command = {key: function_name}

        if payload:
            import json as json_module

            try:
                payload_dict = json_module.loads(payload)
            except (ValueError, TypeError) as e:
                raise ClickException("--payload must be valid JSON: {}".format(e))
            if not isinstance(payload_dict, dict):
                raise ClickException("--payload must be a JSON object (dict), not {}.".format(type(payload_dict).__name__))
            command.update(payload_dict)

        client_context = base64.b64encode(client_context.encode("utf-8")).decode("utf-8") if client_context else None

        # Can't use hjson
        import json as json

        response = self.zappa.invoke_lambda_function(
            self.lambda_name,
            json.dumps(command),
            invocation_type="RequestResponse",
            client_context=client_context,
            qualifier=qualifier,
        )

        print(self.format_lambda_response(response, not no_color))

        # For a successful request FunctionError is not in response.
        # https://github.com/Miserlou/Zappa/pull/1254/
        if "FunctionError" in response:
            raise ClickException("{} error occurred while invoking command.".format(response["FunctionError"]))

    def format_lambda_response(self, response, colorize=True):
        if "LogResult" in response: