Merge pull request #313 from jvanbuel/bugfix/make-state-optional-for-running-dagruns

Make state field optional so running DagRuns can be deserialized correctly

Author: jvanbuel

src/airflow/model/taskinstance.rs

@@ -25,7 +25,7 @@ pub struct TaskInstance {
     #[serde(rename = "end_date", with = "time::serde::iso8601::option")]
     pub end_date: Option<OffsetDateTime>,
     pub duration: Option<f64>,
-    pub state: String,
+    pub state: Option<String>,
     #[serde(rename = "try_number")]
     pub try_number: i64,
     #[serde(rename = "map_index")]

src/app/model/taskinstances.rs

@@ -80,7 +80,7 @@ impl TaskInstanceModel {
     pub fn mark_task_instance(&mut self, task_id: &str, status: &str) {
         self.filtered.items.iter_mut().for_each(|task_instance| {
             if task_instance.task_id == task_id {
-                task_instance.state = status.to_string();
+                task_instance.state = Some(status.to_string());
             }
         });
     }
@@ -295,14 +295,20 @@ impl Widget for &mut TaskInstanceModel {
                 } else {
                     "None".to_string()
                 }),
-                Line::from(match item.state.as_str() {
-                    "success" => state_to_colored_square(AirflowStateColor::Success),
-                    "running" => state_to_colored_square(AirflowStateColor::Running),
-                    "failed" => state_to_colored_square(AirflowStateColor::Failed),
-                    "queued" => state_to_colored_square(AirflowStateColor::Queued),
-                    "up_for_retry" => state_to_colored_square(AirflowStateColor::UpForRetry),
-                    "upstream_failed" => state_to_colored_square(AirflowStateColor::UpstreamFailed),
-                    _ => state_to_colored_square(AirflowStateColor::None),
+                Line::from(if let Some(state) = &item.state {
+                    match state.as_str() {
+                        "success" => state_to_colored_square(AirflowStateColor::Success),
+                        "running" => state_to_colored_square(AirflowStateColor::Running),
+                        "failed" => state_to_colored_square(AirflowStateColor::Failed),
+                        "queued" => state_to_colored_square(AirflowStateColor::Queued),
+                        "up_for_retry" => state_to_colored_square(AirflowStateColor::UpForRetry),
+                        "upstream_failed" => {
+                            state_to_colored_square(AirflowStateColor::UpstreamFailed)
+                        }
+                        _ => state_to_colored_square(AirflowStateColor::None),
+                    }
+                } else {
+                    state_to_colored_square(AirflowStateColor::None)
                 }),
                 Line::from(format!("{:?}", item.try_number)),
             ])

src/app/worker.rs

@@ -169,7 +169,11 @@ impl Worker {
                         app.task_instances.all = task_instances.task_instances;
                         app.task_instances.filter_task_instances();
                     }
-                    Err(e) => app.task_instances.errors.push(e),
+
+                    Err(e) => {
+                        log::error!("Error getting task instances: {:?}", e);
+                        app.task_instances.errors.push(e);
+                    }
                 }
             }
             WorkerMessage::GetDagCode { dag_id } => {