""" Isolates interactive init prompt flow. Expected to call generator logic at end of flow. """ import logging import pathlib import re import tempfile from typing import Optional, Tuple import click from botocore.exceptions import ClientError, WaiterError from samcli.commands._utils.options import generate_next_command_recommendation from samcli.commands.exceptions import InvalidInitOptionException, PopularRuntimeNotFoundException, SchemasApiException from samcli.commands.init.init_flow_helpers import ( _get_image_from_runtime, _get_runtime_from_image, _get_templates_with_dependency_manager, get_architectures, get_sorted_runtimes, ) from samcli.commands.init.init_generator import do_generate from samcli.commands.init.init_templates import InitTemplates, InvalidInitTemplateError from samcli.commands.init.interactive_event_bridge_flow import ( get_schema_template_details, get_schemas_api_caller, get_schemas_template_parameter, ) from samcli.lib.config.samconfig import DEFAULT_CONFIG_FILE_NAME from samcli.lib.schemas.schemas_code_manager import do_download_source_code_binding, do_extract_and_merge_schemas_code from samcli.lib.utils.architecture import SUPPORTED_RUNTIMES from samcli.lib.utils.osutils import remove from samcli.lib.utils.packagetype import IMAGE, ZIP from samcli.local.common.runtime_template import ( LAMBDA_IMAGES_RUNTIMES_MAP, get_provided_runtime_from_custom_runtime, is_custom_runtime, ) LOG = logging.getLogger(__name__) # pylint: disable=too-many-arguments def do_interactive( location, pt_explicit, package_type, runtime, architecture, base_image, dependency_manager, output_dir, name, app_template, no_input, tracing, application_insights, structured_logging, ): """ Implementation of the ``cli`` method when --interactive is provided. It will ask customers a few questions to init a template. """ if app_template: location_opt_choice = "1" else: click.echo("Which template source would you like to use?") click.echo("\t1 - AWS Quick Start Templates\n\t2 - Custom Template Location") location_opt_choice = click.prompt("Choice", type=click.Choice(["1", "2"]), show_choices=False) generate_application( location, pt_explicit, package_type, runtime, architecture, base_image, dependency_manager, output_dir, name, app_template, no_input, location_opt_choice, tracing, application_insights, structured_logging, ) def generate_application( location, pt_explicit, package_type, runtime, architecture, base_image, dependency_manager, output_dir, name, app_template, no_input, location_opt_choice, tracing, application_insights, structured_logging, ): # pylint: disable=too-many-arguments """ The method holds the decision logic for generating an application Parameters ---------- location : str Location to SAM template pt_explicit : bool boolean representing if the customer explicitly stated packageType package_type : str Zip or Image runtime : str AWS Lambda runtime or Custom runtime architecture : str The architecture type 'x86_64' and 'arm64' in AWS base_image : str AWS Lambda base image dependency_manager : str Runtime's Dependency manager output_dir : str Project output directory name : str name of the project app_template : str AWS Serverless Application template no_input : bool Whether to prompt for input or to accept default values (the default is False, which prompts the user for values it doesn't know for baking) location_opt_choice : int User input for selecting how to get customer a vended serverless application tracing : bool boolen value to determine if X-Ray tracing show be activated or not application_insights : bool boolean value to determine if AppInsights monitoring should be enabled or not structured_logging: bool boolean value to determine if Json structured logging should be enabled or not """ if location_opt_choice == "1": _generate_from_use_case( location, pt_explicit, package_type, runtime, base_image, dependency_manager, output_dir, name, app_template, architecture, tracing, application_insights, structured_logging, ) else: _generate_from_location( location, package_type, runtime, dependency_manager, output_dir, name, no_input, tracing, application_insights, structured_logging, ) # pylint: disable=too-many-statements def _generate_from_location( location, package_type, runtime, dependency_manager, output_dir, name, no_input, tracing, application_insights, structured_logging, ): location = click.prompt("\nTemplate location (git, mercurial, http(s), zip, path)", type=str) summary_msg = """ ----------------------- Generating application: ----------------------- Location: {location} Output Directory: {output_dir} """.format( location=location, output_dir=output_dir ) click.echo(summary_msg) do_generate( location, package_type, runtime, dependency_manager, output_dir, name, no_input, None, tracing, application_insights, structured_logging, ) # pylint: disable=too-many-statements def _generate_from_use_case( location: Optional[str], pt_explicit: bool, package_type: Optional[str], runtime: Optional[str], base_image: Optional[str], dependency_manager: Optional[str], output_dir: Optional[str], name: Optional[str], app_template: Optional[str], architecture: Optional[str], tracing: Optional[bool], application_insights: Optional[bool], structured_logging: Optional[bool], ) -> None: templates = InitTemplates() runtime_or_base_image = runtime if runtime else base_image package_type_filter_value = package_type if pt_explicit else None preprocessed_options = templates.get_preprocessed_manifest( runtime_or_base_image, app_template, package_type_filter_value, dependency_manager ) question = "Choose an AWS Quick Start application template" use_case = _get_choice_from_options( None, preprocessed_options, question, "Template", ) default_app_template_properties = _generate_default_hello_world_application( use_case, package_type, runtime, base_image, dependency_manager, pt_explicit ) chosen_app_template_properties = _get_app_template_properties( preprocessed_options, use_case, base_image, default_app_template_properties ) runtime, base_image, package_type, dependency_manager, template_chosen = chosen_app_template_properties if tracing is None: tracing = prompt_user_to_enable_tracing() if application_insights is None: application_insights = prompt_user_to_enable_application_insights() if structured_logging is None: structured_logging = prompt_user_to_enable_structured_logging() app_template = template_chosen["appTemplate"] base_image = ( LAMBDA_IMAGES_RUNTIMES_MAP.get(str(runtime)) if not base_image and package_type == IMAGE else base_image ) if not name: name = click.prompt("\nProject name", type=str, default="sam-app") location = templates.location_from_app_template(package_type, runtime, base_image, dependency_manager, app_template) final_architecture = get_architectures(architecture) lambda_supported_runtime = ( get_provided_runtime_from_custom_runtime(runtime) if is_custom_runtime(runtime) else runtime ) extra_context = { "project_name": name, "runtime": lambda_supported_runtime, "architectures": {"value": final_architecture}, } # executing event_bridge logic if call is for Schema dynamic template is_dynamic_schemas_template = templates.is_dynamic_schemas_template( package_type, app_template, runtime, base_image, dependency_manager ) if is_dynamic_schemas_template: schemas_api_caller = get_schemas_api_caller() schema_template_details = _get_schema_template_details(schemas_api_caller) schemas_template_parameter = get_schemas_template_parameter(schema_template_details) extra_context = {**schemas_template_parameter, **extra_context} no_input = True summary_msg = generate_summary_message( package_type, runtime, base_image, dependency_manager, output_dir, name, app_template, final_architecture ) click.echo(summary_msg) command_suggestions = generate_next_command_recommendation( [ ("Create pipeline", f"cd {name} && sam pipeline init --bootstrap"), ("Validate SAM template", f"cd {name} && sam validate"), ("Test Function in the Cloud", f"cd {name} && sam sync --stack-name {{stack-name}} --watch"), ] ) click.secho(command_suggestions, fg="yellow") do_generate( location, package_type, lambda_supported_runtime, dependency_manager, output_dir, name, no_input, extra_context, tracing, application_insights, structured_logging, ) # executing event_bridge logic if call is for Schema dynamic template if is_dynamic_schemas_template: _package_schemas_code( lambda_supported_runtime, schemas_api_caller, schema_template_details, output_dir, name, location ) def _get_latest_python_runtime() -> str: """ Returns the latest support version of Python SAM CLI supports Returns ------- str: The name of the latest Python runtime (ex. "python3.12") """ latest_major = 0 latest_minor = 0 compiled_regex = re.compile(r"python(.*?)\.(.*)") for runtime in SUPPORTED_RUNTIMES: if not runtime.startswith("python"): continue # python3.12 => 3.12 => (3, 12) version_match = re.match(compiled_regex, runtime) if not version_match: LOG.debug(f"Failed to match version while checking {runtime}") continue matched_groups = version_match.groups() try: version_major = int(matched_groups[0]) version_minor = int(matched_groups[1]) except (ValueError, IndexError): LOG.debug(f"Failed to parse version while checking {runtime}") continue if version_major > latest_major: latest_major = version_major latest_minor = version_minor elif version_major == latest_major: latest_minor = version_minor if version_minor > latest_minor else latest_minor if not latest_major: # major version is still 0, assume that something went wrong # this in theory should not happen as long as Python is # listed in the SUPPORTED_RUNTIMES constant raise PopularRuntimeNotFoundException("Was unable to search for the latest supported runtime") selected_version = f"python{latest_major}.{latest_minor}" LOG.debug(f"Using {selected_version} as the latest runtime version") return selected_version def _generate_default_hello_world_application( use_case: str, package_type: Optional[str], runtime: Optional[str], base_image: Optional[str], dependency_manager: Optional[str], pt_explicit: bool, ) -> Tuple: """ Generate the default Hello World template if Hello World Example is selected Parameters ---------- use_case : str Type of template example selected package_type : Optional[str] The package type, 'Zip' or 'Image', see samcli/lib/utils/packagetype.py runtime : Optional[str] AWS Lambda function runtime base_image : Optional[str] AWS Lambda function base-image dependency_manager : Optional[str] dependency manager pt_explicit : bool True --package-type was passed or Vice versa Returns ------- Tuple configuration for a default Hello World Example """ is_package_type_image = bool(package_type == IMAGE) if use_case == "Hello World Example" and not (runtime or base_image or is_package_type_image or dependency_manager): latest_python = _get_latest_python_runtime() if click.confirm(f"\nUse the most popular runtime and package type? ({latest_python} and zip)"): runtime, package_type, dependency_manager, pt_explicit = _get_latest_python_runtime(), ZIP, "pip", True return (runtime, package_type, dependency_manager, pt_explicit) def _get_app_template_properties( preprocessed_options: dict, use_case: str, base_image: Optional[str], template_properties: Tuple ) -> Tuple: """ This is the heart of the interactive flow, this method fetchs the templates options needed to generate a template Parameters ---------- preprocessed_options : dict Preprocessed manifest from https://github.com/aws/aws-sam-cli-app-templates use_case : Optional[str] Type of template example selected base_image : str AWS Lambda function base-image template_properties : Tuple Tuple of template properties like runtime, packages type and dependency manager Returns ------- Tuple Tuple of template configuration and the chosen template Raises ------ InvalidInitOptionException exception raised when invalid option is provided """ runtime, package_type, dependency_manager, pt_explicit = template_properties runtime_options = preprocessed_options[use_case] runtime = None if is_custom_runtime(runtime) else runtime if not runtime and not base_image: question = "Which runtime would you like to use?" runtime = _get_choice_from_options(runtime, runtime_options, question, "Runtime") if base_image: runtime = _get_runtime_from_image(base_image) if runtime is None: raise InvalidInitOptionException(f"Runtime could not be inferred for base image {base_image}.") package_types_options = runtime_options.get(runtime) if not package_types_options: raise InvalidInitOptionException(f"Lambda Runtime {runtime} is not supported for {use_case} examples.") if not pt_explicit: message = "What package type would you like to use?" package_type = _get_choice_from_options(None, package_types_options, message, "Package type") if package_type == IMAGE: base_image = _get_image_from_runtime(runtime) dependency_manager_options = package_types_options.get(package_type) if not dependency_manager_options: raise InvalidInitOptionException( f"{package_type} package type is not supported for {use_case} examples and runtime {runtime} selected." ) dependency_manager = _get_dependency_manager(dependency_manager_options, dependency_manager, runtime) template_chosen = _get_app_template_choice(dependency_manager_options, dependency_manager) return (runtime, base_image, package_type, dependency_manager, template_chosen) def prompt_user_to_enable_tracing(): """ Prompt user to if X-Ray Tracing should activated for functions in the SAM template and vice versa """ if click.confirm("\nWould you like to enable X-Ray tracing on the function(s) in your application? "): doc_link = "https://aws.amazon.com/xray/pricing/" click.echo(f"X-Ray will incur an additional cost. View {doc_link} for more details") return True return False def prompt_user_to_enable_application_insights(): """ Prompt user to choose if AppInsights monitoring should be enabled for their application and vice versa """ doc_link = "https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/cloudwatch-application-insights.html" if click.confirm( f"\nWould you like to enable monitoring using CloudWatch Application Insights?" f"\nFor more info, please view {doc_link}" ): pricing_link = ( "https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring" "/appinsights-what-is.html#appinsights-pricing" ) click.echo(f"AppInsights monitoring may incur additional cost. View {pricing_link} for more details") return True return False def prompt_user_to_enable_structured_logging(): """ Prompt user to choose if structured loggingConfig should activated for their functions in the SAM template and vice versa """ if click.confirm("\nWould you like to set Structured Logging in JSON format on your Lambda functions? "): doc_link = ( "https://docs.aws.amazon.com/lambda/latest/dg/" "monitoring-cloudwatchlogs.html#monitoring-cloudwatchlogs-pricing" ) click.echo( f"Structured Logging in JSON format might incur an additional cost. View {doc_link} for more details" ) return True return False def _get_choice_from_options(chosen, options, question, msg): if chosen: return chosen click_choices = [] options_list = options if isinstance(options, list) else list(options.keys()) options_list = get_sorted_runtimes(options_list) if msg == "Runtime" else options_list if not options_list: raise InvalidInitOptionException(f"There are no {msg} options available to be selected.") if len(options_list) == 1: click.echo( f"\nBased on your selections, the only {msg} available is {options_list[0]}." + f"\nWe will proceed to selecting the {msg} as {options_list[0]}." ) return options_list[0] click.echo(f"\n{question}") for index, option in enumerate(options_list): click.echo(f"\t{index+1} - {option}") click_choices.append(str(index + 1)) choice = click.prompt(msg, type=click.Choice(click_choices), show_choices=False) return options_list[int(choice) - 1] def _get_app_template_choice(templates_options, dependency_manager): templates = _get_templates_with_dependency_manager(templates_options, dependency_manager) chosen_template = templates[0] if len(templates) > 1: click.echo("\nSelect your starter template") click_template_choices = [] for index, template in enumerate(templates): click.echo(f"\t{index+1} - {template['displayName']}") click_template_choices.append(str(index + 1)) template_choice = click.prompt("Template", type=click.Choice(click_template_choices), show_choices=False) chosen_template = templates[int(template_choice) - 1] return chosen_template def _get_dependency_manager(options, dependency_manager, runtime): valid_dep_managers = sorted(list(set(template["dependencyManager"] for template in options))) if not dependency_manager: if len(valid_dep_managers) == 1: dependency_manager = valid_dep_managers[0] click.echo( f"\nBased on your selections, the only dependency manager available is {dependency_manager}." + f"\nWe will proceed copying the template using {dependency_manager}." ) else: question = "Which dependency manager would you like to use?" dependency_manager = _get_choice_from_options( dependency_manager, valid_dep_managers, question, "Dependency manager" ) elif dependency_manager and dependency_manager not in valid_dep_managers: msg = ( f"Lambda Runtime {runtime} and dependency manager {dependency_manager} " + "do not have an available initialization template." ) raise InvalidInitTemplateError(msg) return dependency_manager def _get_schema_template_details(schemas_api_caller): try: return get_schema_template_details(schemas_api_caller) except ClientError as e: raise SchemasApiException( "Exception occurs while getting Schemas template parameter. %s" % e.response["Error"]["Message"] ) from e def _package_schemas_code(runtime, schemas_api_caller, schema_template_details, output_dir, name, location): try: click.echo("Trying to get package schema code") with tempfile.NamedTemporaryFile(delete=False) as download_location: do_download_source_code_binding(runtime, schema_template_details, schemas_api_caller, download_location) do_extract_and_merge_schemas_code(download_location, output_dir, name, location) except (ClientError, WaiterError) as e: raise SchemasApiException( "Exception occurs while packaging Schemas code. %s" % e.response["Error"]["Message"] ) from e finally: remove(download_location.name) def generate_summary_message( package_type, runtime, base_image, dependency_manager, output_dir, name, app_template, architecture ): """ Parameters ---------- package_type : str The package type, 'Zip' or 'Image', see samcli/lib/utils/packagetype.py runtime : str AWS Lambda function runtime base_image : str base image dependency_manager : str dependency manager output_dir : str the directory where project will be generated in name : str Project Name app_template : str application template generated architecture : list Architecture type either x86_64 or arm64 on AWS lambda Returns ------- str Summary Message of the application template generated """ summary_msg = "" if package_type == ZIP: summary_msg = f""" ----------------------- Generating application: ----------------------- Name: {name} Runtime: {runtime} Architectures: {architecture[0]} Dependency Manager: {dependency_manager} Application Template: {app_template} Output Directory: {output_dir} Configuration file: {pathlib.Path(output_dir).joinpath(name, DEFAULT_CONFIG_FILE_NAME)} Next steps can be found in the README file at {pathlib.Path(output_dir).joinpath(name, "README.md")} """ elif package_type == IMAGE: summary_msg = f""" ----------------------- Generating application: ----------------------- Name: {name} Base Image: {base_image} Architectures: {architecture[0]} Dependency Manager: {dependency_manager} Output Directory: {output_dir} Configuration file: {pathlib.Path(output_dir).joinpath(name, DEFAULT_CONFIG_FILE_NAME)} Next steps can be found in the README file at {pathlib.Path(output_dir).joinpath(name, "README.md")} """ return summary_msg