def validate_task()

in scripts/generate_dag.py [0:0]

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

def validate_task(task: dict, airflow_version: str):
    if not task.get("operator"):
        raise KeyError(f"`operator` key must exist in {task}")

    if not task["operator"] in AIRFLOW_IMPORTS[airflow_version]:
        raise ValueError(
            f"`task.operator` must be one of {list(AIRFLOW_IMPORTS[airflow_version].keys())}"
        )

    if not task["args"].get("task_id"):
        raise KeyError(f"`args.task_id` key must exist in {task}")