def read_config_file()

in src/datamigration/dags/datamigration_utils/hive_dvt_check_utils.py [0:0]

• 34 lines of code
• 1 McCabe index (conditional complexity)

def read_config_file(translation_config):
    """
    Set required variables as dictionary variables from translation config
    """
    config_dict = {}
    config_dict["rerun_flag"] = translation_config["transfer_config"][
        "rerun_flag"
    ].upper()
    config_dict["hive_db_name"] = translation_config["transfer_config"]["params"][
        "hive_db_name"
    ]
    config_dict["dvt_check_flag"] = translation_config["dvt_check"].upper()
    config_dict["temp_bucket"] = translation_config["transfer_config"]["params"][
        "gcs_temp_bucket"
    ]
    config_dict["bq_dataset"] = translation_config["transfer_config"]["params"][
        "bq_dataset_id"
    ]
    config_dict["hive_gcs_staging_bucket_id"] = translation_config["transfer_config"][
        "params"
    ]["hive_gcs_staging_bucket_id"]
    config_dict["hive_gcs_staging_path"] = translation_config["transfer_config"][
        "params"
    ]["hive_gcs_staging_path"]
    config_dict["project_id"] = translation_config["transfer_config"]["params"][
        "project_id"
    ]
    config_dict["validation_mode"] = translation_config["validation_config"][
        "validation_mode"
    ]
    config_dict["batch_distribution"] = translation_config["batchDistribution"]
    config_dict["hive_ddl_metadata"] = "hive_ddl_metadata"
    config_dict["bq_dataset_audit"] = "dmt_logs"
    config_dict["bq_load_audit"] = "hive_bqload_audit"
    config_dict["dvt_results"] = "dmt_dvt_results"
    config_dict["schema_results_tbl"] = "dmt_schema_results"
    return config_dict