def read_translation_config()

in src/translation/scripts/hive/extract_hive_ddls_manual.py [0:0]

• 32 lines of code
• 4 McCabe index (conditional complexity)

def read_translation_config(translation_config):
    """
    Convert JSON config to dictionary
    """
    dict = {}
    dict["bigquery_audit_table"] = "hive_ddl_metadata"
    dict["bq_dataset_audit"] = "dmt_logs"
    dict["columns"] = [
        "run_id",
        "start_time",
        "database",
        "bq_dataset",
        "table",
        "field_delimiter",
        "partition_flag",
        "cluster_flag",
        "format",
        "ddl_extracted",
    ]
    dict["host_ip"] = translation_config["hive_config"]["server_config"]["connection"][
        "host"
    ]
    source_path = translation_config["migrationTask"]["translationConfigDetails"][
        "gcsSourcePath"
    ]
    dict["bucket_name"] = source_path.split("/")[2]
    dict["gcs_ddl_output_path"] = source_path.split("/", 3)[-1]
    nm_map_list = translation_config["migrationTask"]["translationConfigDetails"][
        "nameMappingList"
    ]["name_map"]
    dict["hive_db"] = list(set(d["source"]["schema"] for d in nm_map_list))[0]
    dict["bq_dataset"] = list(set(d["target"]["schema"] for d in nm_map_list))[0]
    input_tables = translation_config["source_ddl_extract_table_list"]
    dict["input_tables_list"] = [x.lower() for x in input_tables.split(",")]
    return dict