def airflow_task_python_imports_build()

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

• 20 lines of code
• 8 McCabe index (conditional complexity)

def airflow_task_python_imports_build(task, template):
    # Load the Template Output Structure
    if template["target"] is None:
        raise ValueError(
            f"dagify: no target data in template: {template['metadata']['name']}, python imports can not be calculated")
    if template["target"]['operator'] is None:
        raise ValueError(
            f"dagify: no target operator data in template: {template['metadata']['name']}, python imports can not be calculated")
    if template["target"]['operator']['imports'] is None:
        raise ValueError(
            f"dagify: no target operator import data in template: {template['metadata']['name']}, python imports can not be calculated")

    python_imports = []
    for imp in template["target"]['operator']['imports']:
        if imp['package'] is None:
            continue
        if imp['imports'] is None:
            continue
        if len(imp['imports']) is None:
            continue
        python_imports.append(imp)
    return python_imports