feat: Add click command for cloud invoke command

samcli/cli/command.py

@@ -30,6 +30,7 @@
     "samcli.commands.pipeline.pipeline",
     "samcli.commands.list.list",
     "samcli.commands.docs",
+    # "samcli.commands.cloud.cloud",
     # We intentionally do not expose the `bootstrap` command for now. We might open it up later
     # "samcli.commands.bootstrap",
 ]

samcli/cli/types.py

@@ -3,6 +3,7 @@
 """
 
 import json
+import logging
 import re
 from json import JSONDecodeError
 
@@ -12,6 +13,8 @@
 
 PARAM_AND_METADATA_KEY_REGEX = """([A-Za-z0-9\\"\']+)"""
 
+LOG = logging.getLogger(__name__)
+
 
 def _generate_match_regex(match_pattern, delim):
     """
@@ -421,3 +424,57 @@ def convert(self, value, param, ctx):
         if not is_ecr_url(_value):
             raise click.BadParameter(f"{param.opts[0]} needs to have valid ECR URI as value")
         return {key: _value}
+
+
+class RemoteInvokeBotoApiParameterType(click.ParamType):
+    """
+    Custom Parameter Type for Multi valued Boto API parameter option of remote invoke command.
+    """
+
+    name = ""
+    MIN_KEY_VALUE_PAIR_LENGTH = 2
+
+    def convert(self, value, param, ctx):
+        """Converts the user provided parameter value with the format "parameter=value" to dict
+            {"parameter": "value"}
+
+        Parameters
+        ------------
+        value: User provided value for the click option
+        param: click parameter
+        ctx: Context
+        """
+        # Split by first "=" as some values could have multiple "=" For e.g. base-64 encoded ClientContext for Lambda
+        key_value_pair = value.split("=", 1)
+        if len(key_value_pair) < self.MIN_KEY_VALUE_PAIR_LENGTH:
+            raise click.BadParameter(
+                f"{param.opts[0]} is not a valid format, it needs to be of the form parameter_key=parameter_value"
+            )
+        key = key_value_pair[0]
+        _value = key_value_pair[1]
+        LOG.debug("Converting provided %s option value to dict", param.opts[0])
+        return {key: _value}
+
+
+class RemoteInvokeOutputFormatType(click.Choice):
+    """
+    Custom Parameter Type for output-format option of remote invoke command.
+    """
+
+    def __init__(self, enum):
+        self.enum = enum
+        super().__init__(choices=[item.name.lower() for item in enum])
+
+    def convert(self, value, param, ctx):
+        """Converts the user provided parameter value for the option to
+           the provided Enum
+
+        Parameters
+        ------------
+        value: User provided value for the click option
+        param: click parameter
+        ctx: Context
+        """
+        LOG.debug("Converting provided %s option value to Enum", param.opts[0])
+        value = super().convert(value, param, ctx)
+        return self.enum(value)

samcli/commands/_utils/options.py

@@ -16,6 +16,7 @@
     CfnTags,
     ImageRepositoriesType,
     ImageRepositoryType,
+    RemoteInvokeBotoApiParameterType,
     SigningProfilesOptionType,
 )
 from samcli.commands._utils.constants import (
@@ -129,6 +130,21 @@ def image_repositories_callback(ctx, param, provided_value):
     return image_repositories if image_repositories else None
 
 
+def remote_invoke_boto_parameter_callback(ctx, param, provided_value):
+    """
+    Create an dictionary of boto parameters to their provided values.
+    :param ctx: Click Context
+    :param param: Param name
+    :param provided_value: Value provided by Click, after being processed by RemoteInvokeBotoApiParameterType.
+    :return: dictionary of boto api parameters to their provided values. E.g. LogType=Tail for Lambda invoke API
+    """
+    boto_api_parameters = {}
+    for value in provided_value:
+        boto_api_parameters.update(value)
+
+    return boto_api_parameters
+
+
 def artifact_callback(ctx, param, provided_value, artifact):
     """
     Provide an error if there are zip/image artifact based resources,
@@ -571,6 +587,21 @@ def image_repositories_option(f):
     return image_repositories_click_option()(f)
 
 
+def remote_invoke_parameter_click_option():
+    return click.option(
+        "--parameter",
+        multiple=True,
+        type=RemoteInvokeBotoApiParameterType(),
+        callback=remote_invoke_boto_parameter_callback,
+        required=False,
+        help="Additional parameters for the boto API call.\n" "Lambda APIs: invoke and invoke_with_response_stream",
+    )
+
+
+def remote_invoke_parameter_option(f):
+    return remote_invoke_parameter_click_option()(f)
+
+
 def s3_prefix_click_option():
     return click.option(
         "--s3-prefix",

samcli/commands/cloud/cloud.py

@@ -0,0 +1,19 @@
+"""
+Command group for "cloud" suite for commands. It provides common CLI arguments for interacting with
+cloud resources such as Lambda Function.
+"""
+
+import click
+
+from samcli.commands.cloud.invoke.cli import cli as invoke_cli
+
+
+@click.group()
+def cli():
+    """
+    Interact with your Serverless application in the cloud for quick development & testing
+    """
+
+
+# Add individual commands under this group
+cli.add_command(invoke_cli)

samcli/commands/cloud/invoke/cli.py

@@ -0,0 +1,139 @@
+"""CLI command for "invoke" command."""
+import logging
+from io import TextIOWrapper
+from typing import cast
+
+import click
+
+from samcli.cli.cli_config_file import TomlProvider, configuration_option
+from samcli.cli.context import Context
+from samcli.cli.main import aws_creds_options, common_options, pass_context, print_cmdline_args
+from samcli.cli.types import RemoteInvokeOutputFormatType
+from samcli.commands._utils.options import remote_invoke_parameter_option
+from samcli.lib.cli_validation.remote_invoke_options_validations import (
+    event_and_event_file_options_validation,
+    stack_name_or_resource_id_atleast_one_option_validation,
+)
+from samcli.lib.remote_invoke.remote_invoke_executors import RemoteInvokeOutputFormat
+from samcli.lib.telemetry.metric import track_command
+from samcli.lib.utils.version_checker import check_newer_version
+
+LOG = logging.getLogger(__name__)
+
+HELP_TEXT = """
+Invoke or send an event to cloud resources in your CFN stack
+"""
+SHORT_HELP = "Invoke a deployed resource in the cloud"
+
+
+@click.command("invoke", help=HELP_TEXT, short_help=SHORT_HELP)
+@configuration_option(provider=TomlProvider(section="parameters"))
+@click.option("--stack-name", required=False, help="Name of the stack to get the resource information from")
+@click.option("--resource-id", required=False, help="Name of the resource that will be invoked")
+@click.option(
+    "--event",
+    "-e",
+    help="The event that will be sent to the resource. The target parameter will depend on the resource type. "
+    "For instance: 'Payload' for Lambda",
+)
+@click.option(
+    "--event-file",
+    type=click.File("r", encoding="utf-8"),
+    help="The file that contains the event that will be sent to the resource",
+)
+@click.option(
+    "--output-format",
+    help="Output format for the boto API response",
+    default=RemoteInvokeOutputFormat.DEFAULT.name.lower(),
+    type=RemoteInvokeOutputFormatType(RemoteInvokeOutputFormat),
+)
+@remote_invoke_parameter_option
+@stack_name_or_resource_id_atleast_one_option_validation
+@event_and_event_file_options_validation
+@common_options
+@aws_creds_options
+@pass_context
+@track_command
+@check_newer_version
+@print_cmdline_args
+def cli(
+    ctx: Context,
+    stack_name: str,
+    resource_id: str,
+    event: str,
+    event_file: TextIOWrapper,
+    output_format: RemoteInvokeOutputFormat,
+    parameter: dict,
+    config_file: str,
+    config_env: str,
+) -> None:
+    """
+    `sam cloud invoke` command entry point
+    """
+
+    do_cli(
+        stack_name,
+        resource_id,
+        event,
+        event_file,
+        output_format,
+        parameter,
+        ctx.region,
+        ctx.profile,
+        config_file,
+        config_env,
+    )
+
+
+def do_cli(
+    stack_name: str,
+    resource_id: str,
+    event: str,
+    event_file: TextIOWrapper,
+    output_format: RemoteInvokeOutputFormat,
+    parameter: dict,
+    region: str,
+    profile: str,
+    config_file: str,
+    config_env: str,
+) -> None:
+    """
+    Implementation of the ``cli`` method
+    """
+    from samcli.commands.cloud.remote_invoke_context import RemoteInvokeContext
+    from samcli.commands.exceptions import UserException
+    from samcli.lib.remote_invoke.exceptions import (
+        ErrorBotoApiCallException,
+        InvalideBotoResponseException,
+        InvalidResourceBotoParameterException,
+    )
+    from samcli.lib.remote_invoke.remote_invoke_executors import RemoteInvokeExecutionInfo
+    from samcli.lib.utils.boto_utils import get_boto_client_provider_with_config, get_boto_resource_provider_with_config
+
+    boto_client_provider = get_boto_client_provider_with_config(region_name=region)
+    boto_resource_provider = get_boto_resource_provider_with_config(region_name=region)
+    try:
+        with RemoteInvokeContext(
+            boto_client_provider=boto_client_provider,
+            boto_resource_provider=boto_resource_provider,
+            stack_name=stack_name,
+            resource_id=resource_id,
+        ) as remote_invoke_context:
+
+            remote_invoke_input = RemoteInvokeExecutionInfo(
+                payload=event, payload_file=event_file, parameters=parameter, output_format=output_format
+            )
+
+            remote_invoke_result = remote_invoke_context.run(remote_invoke_input=remote_invoke_input)
+
+            if remote_invoke_result.is_succeeded():
+                LOG.debug("Invoking resource was successfull, writing response to stdout")
+                if remote_invoke_result.log_output:
+                    LOG.debug("Writing log output to stderr")
+                    remote_invoke_context.stderr.write(remote_invoke_result.log_output.encode())
+                output_response = cast(str, remote_invoke_result.response)
+                remote_invoke_context.stdout.write(output_response.encode())
+            else:
+                raise cast(Exception, remote_invoke_result.exception)
+    except (ErrorBotoApiCallException, InvalideBotoResponseException, InvalidResourceBotoParameterException) as ex:
+        raise UserException(str(ex), wrapped_from=ex.__class__.__name__) from ex

samcli/commands/remote_invoke/remote_invoke_context.py → samcli/commands/cloud/remote_invoke_context.py

@@ -4,7 +4,7 @@
 import logging
 from typing import Optional, cast
 
-from samcli.commands.remote_invoke.exceptions import (
+from samcli.commands.cloud.exceptions import (
     AmbiguousResourceForRemoteInvoke,
     InvalidRemoteInvokeParameters,
     NoExecutorFoundForRemoteInvoke,
@@ -13,6 +13,7 @@
 )
 from samcli.lib.remote_invoke.remote_invoke_executor_factory import RemoteInvokeExecutorFactory
 from samcli.lib.remote_invoke.remote_invoke_executors import RemoteInvokeExecutionInfo
+from samcli.lib.utils import osutils
 from samcli.lib.utils.arn_utils import ARNParts, InvalidArnValue
 from samcli.lib.utils.boto_utils import BotoProviderType
 from samcli.lib.utils.cloudformation import (
@@ -22,6 +23,7 @@
     get_resource_summary_from_physical_id,
 )
 from samcli.lib.utils.resources import AWS_LAMBDA_FUNCTION
+from samcli.lib.utils.stream_writer import StreamWriter
 
 LOG = logging.getLogger(__name__)
 
@@ -186,3 +188,29 @@ def _get_from_physical_resource_id(self) -> CloudFormationResourceSummary:
                     f"Please provide full resource ARN or --stack-name to resolve the ambiguity."
                 )
             return resource_summary
+
+    @property
+    def stdout(self) -> StreamWriter:
+        """
+        Returns stream writer for stdout to output Lambda function logs to
+
+        Returns
+        -------
+        samcli.lib.utils.stream_writer.StreamWriter
+            Stream writer for stdout
+        """
+        stream = osutils.stdout()
+        return StreamWriter(stream, auto_flush=True)
+
+    @property
+    def stderr(self) -> StreamWriter:
+        """
+        Returns stream writer for stderr to output Lambda function errors to
+
+        Returns
+        -------
+        samcli.lib.utils.stream_writer.StreamWriter
+            Stream writer for stderr
+        """
+        stream = osutils.stderr()
+        return StreamWriter(stream, auto_flush=True)