Fix: An individual plan DAG failure shouldn't prevent the remaining DAGs from loading in Airflow

Author: izeigerman

sqlmesh/schedulers/airflow/dag_generator.py

@@ -90,8 +90,12 @@ def generate_cadence_dags(self, snapshots: t.Iterable[SnapshotIdLike]) -> t.List
                 dags.append(self._create_cadence_dag_for_snapshot(snapshot, snapshots))
         return dags
 
-    def generate_plan_application_dag(self, spec: common.PlanDagSpec) -> DAG:
-        return self._create_plan_application_dag(spec)
+    def generate_plan_application_dag(self, spec: common.PlanDagSpec) -> t.Optional[DAG]:
+        try:
+            return self._create_plan_application_dag(spec)
+        except Exception:
+            logger.exception("Failed to generate the plan application DAG '%s'", spec.request_id)
+            return None
 
     def _create_cadence_dag_for_snapshot(
         self, snapshot: Snapshot, snapshots: t.Dict[SnapshotId, Snapshot]

sqlmesh/schedulers/airflow/integration.py

@@ -144,7 +144,7 @@ def dags(self) -> t.List[DAG]:
                 self._create_janitor_dag(),
             ]
 
-            return system_dags + cadence_dags + plan_application_dags
+            return system_dags + cadence_dags + [d for d in plan_application_dags if d]
 
     def _create_janitor_dag(self) -> DAG:
         dag = self._create_system_dag(common.JANITOR_DAG_ID, self._janitor_interval)