def get_object_to_ingest()

in src/data_load/load.py [0:0]

• 31 lines of code
• 11 McCabe index (conditional complexity)

def get_object_to_ingest(file_location_map, standard_reference, filepath): 
    logger.debug(f"parsing file for ingestion - {filepath}")

    if filepath.endswith(".json"):
        with open(filepath) as file:
            manifest_file = json.load(file)

            # For work product manifests inject the proper data-partition-id
            if standard_reference:
                a_data_partition = config.get("CONNECTION", "data-partition-id")
                new_manifest = json.dumps(manifest_file).replace('{{NAMESPACE}}', a_data_partition)
                manifest_file = json.loads(new_manifest)

    else:
        return None, None

    if not manifest_file:
        logger.error(f"Error with file {filepath}. File is empty.")
    elif "ReferenceData" in manifest_file and len(manifest_file["ReferenceData"]) > 0:
        object_to_ingest = update_reference_data_metadata(
            manifest_file["ReferenceData"], standard_reference)
        data_type = "ReferenceData"
    elif "MasterData" in manifest_file and len(manifest_file["MasterData"]) > 0:
        object_to_ingest = add_metadata(manifest_file["MasterData"])
        data_type = "MasterData"
    elif "Data" in manifest_file:
        data_type = "Data"
        if file_location_map is None or len(file_location_map) == 0:
            raise Exception(
                'File Location Map file path is required for Work-Product data ingestion')
        object_to_ingest = update_work_products_metadata(manifest_file["Data"], file_location_map,
                                                         get_directory_name(filepath))
    else:
        object_to_ingest = None
        data_type = None
    return object_to_ingest, data_type