Fix: do not throw error if airflow api failed

Author: sagargg

ckanext/aircan/logic/action.py

@@ -44,24 +44,25 @@ def aircan_submit(context, data_dict: Dict[str, Any]) -> Dict[str, Any]:
     client = AirflowClient()
     try:
         dag_run = client.trigger_dag(conf=payload)
+        context.update({"session": context["model"].meta.create_local_session()})
+        tk.get_action("aircan_status_update")(
+            context,
+            {
+                "resource_id": data_dict.get("id"),
+                "dag_run_id": dag_run.get("dag_run_id"),
+                "state": "queued",
+                "message": f"Added to the queue to be processed with '{dag_run.get("dag_run_id")}'.",
+                "clear_logs": True,
+            },
+        )
+        return {
+            "dag_run": dag_run,
+            "dag_run_id": dag_run.get("dag_run_id"),
+        }
     except requests.HTTPError as e:
         log.error(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e))
 
-    context.update({"session": context["model"].meta.create_local_session()})
-    tk.get_action("aircan_status_update")(
-        context,
-        {
-            "resource_id": data_dict.get("id"),
-            "dag_run_id": dag_run.get("dag_run_id"),
-            "state": "queued",
-            "message": f"Added to the queue to be processed with '{dag_run.get("dag_run_id")}'.",
-            "clear_logs": True,
-        },
-    )
-    return {
-        "dag_run": dag_run,
-        "dag_run_id": dag_run.get("dag_run_id"),
-    }
+    
 
 
 def aircan_status(context, data_dict: Dict[str, Any]) -> Dict[str, Any]: