def get_jobtypes_andcount()

in dagify/converter/utils.py [0:0]

• 21 lines of code
• 10 McCabe index (conditional complexity)

def get_jobtypes_andcount(source_path):
    """Generic function that calculates the job_types and the count from any input"""
    unique_job_types = []
    job_types_source = []
    job_types_count = 0
    if source_path.endswith('.xml'):
        tree = ET.parse(source_path)
        root = tree.getroot()
        # Find all JOB elements
        job_elements = root.findall('.//JOB')
        # Extract TASKTYPE values and store them in a set to ensure uniqueness
        job_types_source = list({job.get('TASKTYPE') for job in job_elements})
        # Convert all to lowercase for comparision
        job_types_source = [item.lower() for item in job_types_source]
        unique_job_types = list(set(job_types_source))
        job_types_count = len(unique_job_types)
    elif source_path.endswith('config.yaml'):
        with open(source_path, 'r') as file:
            data = yaml.safe_load(file)
        for mapping in data['config']['mappings']:
            job_types_source.append(mapping['job_type'])

        # Convert all to lowercase for comparision
        job_types_source = [item.lower() for item in job_types_source]
        unique_job_types = list(set(job_types_source))
        job_types_count = len(unique_job_types)
    return unique_job_types, job_types_count