def process_component()

in solutions_builder/cli/component.py [0:0]

• 67 lines of code
• 14 McCabe index (conditional complexity)

def process_component(method,
                      component_name,
                      template_path,
                      solution_path,
                      destination_path,
                      data={},
                      use_existing_answers=False):

  assert template_path, "template_path is empty."
  current_dir = os.path.dirname(__file__)
  template_dir = f"{current_dir}/../modules/{template_path}"
  answers_file = None

  # Get basic info from root sb.yaml.
  sb_yaml = read_yaml(f"{solution_path}/sb.yaml")
  global_variables = sb_yaml.get("global_variables", {})
  component_answers = {}

  # If the component name is a Git URL, use the URL as-is in copier.
  if check_git_url(template_path):
    template_dir = clone_remote_git(template_path)

  # Otherwise, try to locate the component in local modules/ folder.
  else:
    if method == "update":
      data["component_name"] = component_name
      if component_name not in sb_yaml["components"]:
        raise NameError(
            f"Component {component_name} is not defined in the root yaml 'sb.yaml' file."
        )
      component_answers = sb_yaml["components"][component_name]
      template_path = component_answers["template_path"]
      answers_file = f".sb/module_answers/{component_name}.yaml"

      # Use existing answer values in data, skipping the prompt.
      if use_existing_answers:
        answers_yaml = read_yaml(answers_file)
        for key, value in answers_yaml.items():
          data[key] = value

    else:
      template_dir = f"{current_dir}/../modules/{template_path}"
      if not os.path.exists(template_dir):
        # If module does not exist in the solutions-builder package,
        # try loading from the local file path.
        template_dir = template_path
        if not os.path.exists(template_dir):
          raise FileNotFoundError(
              f"Component '{template_path}' does not exist.")

    # # Get destination_path defined in copier.yaml
    # destination_path = solution_path + "/" + copier_dict["_metadata"].get(
    #     "destination_path")
    # destination_path = destination_path.replace("//", "/")

  copier_dict = get_copier_yaml(template_dir)
  data["component_name"] = component_name
  if "project_id" in data:
    data["project_id"] = global_variables.get("project_id")
  if "project_number" in data:
    data["project_number"] = global_variables.get("project_number")
  data["solution_path"] = solution_path
  data["template_path"] = template_path

  # Run copier with data.
  worker = run_copy(template_dir,
                    ".",
                    data=data,
                    answers_file=answers_file,
                    unsafe=True)

  # Get answer values inputed by user.
  answers = worker.answers.user

  for key, value in worker.answers.user_defaults.items():
    if key not in answers:
      answers[key] = component_answers.get(key) or value
  answers["template_path"] = template_path
  answers["destination_path"] = destination_path

  # Update component's answer back to sb.yaml.
  update_root_yaml(component_name,
                   answers,
                   solution_path)

  # Patch skaffold.yaml
  for patch_file in copier_dict.get("_patch", []):
    print(f"Patching {patch_file}...")
    new_yaml = patch_yaml(f"{solution_path}/{patch_file}",
                          f"{solution_path}/{patch_file}.patch")
    new_yaml["requires"] = dedupe(new_yaml.get("requires"))
    write_yaml(f"{solution_path}/{patch_file}", new_yaml)
    os.remove(f"{solution_path}/{patch_file}.patch")

  print()