def _parse_args()

in src/common/materializer/generate_build_files.py [0:0]

• 94 lines of code
• 12 McCabe index (conditional complexity)

def _parse_args() -> tuple[str, str, str, str, dict, bool, bool]:
    """Parses arguments and sets up logging."""
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--module_name",
        type=str,
        required=True,
        help="Module for which output dataset is created. Required.")
    parser.add_argument(
        "--target_dataset_type",
        type=str,
        required=False,
        help=("Target dataset type (CDC/Reporting). Must correspond to "
              "what is available under \"dataset\" section in config.json. "
              "Default is \"Reporting\"."),
        default="Reporting")
    parser.add_argument(
        "--config_file",
        type=str,
        required=True,
        help=("Config file containing deployment configurations, "
              "with relative path. Required."))
    parser.add_argument(
        "--materializer_settings_file",
        type=str,
        required=True,
        help="Materializer settings file, with relative path. Required.")
    parser.add_argument(
        "--k9_manifest_file",
        type=str,
        required=False,
        default=None,
        help="K9 Manifest file, with relative path. Required only if "
        "local K9 is being utilized.")
    parser.add_argument(
        "--private_worker_pool",
        type=str,
        required=False,
        default=False,
        help="If customer uses a private worker pool. Optional.")
    parser.add_argument(
        "--debug",
        default=False,
        action="store_true",
        help="Flag to set log level to DEBUG. Default is WARNING")

    args = parser.parse_args()

    enable_debug = args.debug
    logging.basicConfig(level=logging.DEBUG if enable_debug else logging.INFO)

    module_name = args.module_name
    config_file = args.config_file
    materializer_settings_file = args.materializer_settings_file
    target_dataset_type = args.target_dataset_type
    k9_manifest_file = args.k9_manifest_file
    private_worker_pool = bool(args.private_worker_pool)

    logging.info("Arguments:")
    logging.info("  module_name = %s", module_name)
    logging.info("  config_file = %s", config_file)
    logging.info("  target_dataset_type = %s", target_dataset_type)
    logging.info("  materializer_settings_file = %s",
                 materializer_settings_file)
    logging.info("  k9_manifest_file = %s", k9_manifest_file)
    logging.info("  debug = %s", enable_debug)

    # validate arguments.
    # Module Name
    if module_name not in constants.CORTEX_MODULES:
        raise ValueError(
            f"🛑 Invalid module name '{module_name}'. Supported modules are : "
            f"'{constants.CORTEX_MODULES}' (case sensitive).")

    # Target Dataset Type
    lower_tgt_dataset_type = target_dataset_type.lower().replace(" ", "_")

    if lower_tgt_dataset_type not in _CORTEX_DATASET_TYPES_LOWER:
        raise ValueError(
            f"🛑 Invalid target dataset type '{target_dataset_type}'. "
            f"Supported types are : '{_CORTEX_DATASET_TYPES_LOWER}' (case "
            "insensitive).")

    if module_name == "k9" and lower_tgt_dataset_type == "cdc":
        raise ValueError(
            "🛑 Module k9 does not support target dataset type 'cdc'.")

    if lower_tgt_dataset_type == "processing" and module_name != "k9":
        raise ValueError(
            f"🛑 Module '{module_name}' does not support target dataset type "
            "'processing'.")

    # Config file
    if not Path(config_file).is_file():
        raise FileNotFoundError(
            f"🛑 Config file '{config_file}' does not exist.")

    # Settings file
    if not Path(materializer_settings_file).is_file():
        raise FileNotFoundError(
            f"🛑 {target_dataset_type} Settings file "
            f"'{materializer_settings_file}' does not exist.")

    # K9 Manifest file
    try:
        k9_manifest = (k9_deployer.load_k9s_manifest(k9_manifest_file)
                       if k9_manifest_file else {})
    except Exception as e:
        raise FileNotFoundError(f"The specified K9 manifest {k9_manifest_file} "
                                "does not exist.") from e

    return (module_name, target_dataset_type, config_file,
            materializer_settings_file, k9_manifest, private_worker_pool,
            enable_debug)