From a13154fa37b6a444182229dd8e618b205bd8fea4 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 25 Jul 2026 10:55:29 +0000 Subject: [PATCH] Collapse single-implementation Airflow trait layer src/airflow/traits/ defined six operation traits plus an AirflowClient super-trait, all implemented solely by FlowrsClient. The module doc said the traits existed so "different API versions (v1 for Airflow v2, v2 for Airflow v3)" could each implement them, but that split was instead built inside FlowrsClient as an enum, so the abstraction never gained a second implementer. Convert the six impl blocks into inherent impls on FlowrsClient, replace Arc with Arc at every call site, and delete the traits module. The dyn indirection was type erasure over one concrete type, including in tests/common/mod.rs, so no polymorphism is lost. Drops the now-unused async-trait dependency from flowrs-tui; the client crate still uses it for AuthProvider, which has seven implementations. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0127jzUgqqy8JVX5RydfQxh8 --- Cargo.lock | 1 - Cargo.toml | 1 - src/airflow/client/impls/dag_ops.rs | 13 +++---- src/airflow/client/impls/dagrun_ops.rs | 13 +++---- src/airflow/client/impls/dagstats_ops.rs | 7 ++-- src/airflow/client/impls/log_ops.rs | 7 ++-- src/airflow/client/impls/task_ops.rs | 7 ++-- src/airflow/client/impls/taskinstance_ops.rs | 13 +++---- src/airflow/client/mod.rs | 7 ++-- src/airflow/mod.rs | 1 - src/airflow/traits/dag.rs | 20 ----------- src/airflow/traits/dagrun.rs | 25 ------------- src/airflow/traits/dagstats.rs | 11 ------ src/airflow/traits/log.rs | 17 --------- src/airflow/traits/mod.rs | 37 -------------------- src/airflow/traits/task.rs | 11 ------ src/airflow/traits/taskinstance.rs | 37 -------------------- src/app/state/environment_state.rs | 6 ++-- src/app/worker/browser.rs | 10 ++---- src/app/worker/dagruns.rs | 10 +++--- src/app/worker/dags.rs | 10 +++--- src/app/worker/logs.rs | 4 +-- src/app/worker/taskinstances.rs | 8 ++--- src/app/worker/tasks.rs | 10 ++---- tests/common/mod.rs | 5 ++- 25 files changed, 51 insertions(+), 240 deletions(-) delete mode 100644 src/airflow/traits/dag.rs delete mode 100644 src/airflow/traits/dagrun.rs delete mode 100644 src/airflow/traits/dagstats.rs delete mode 100644 src/airflow/traits/log.rs delete mode 100644 src/airflow/traits/mod.rs delete mode 100644 src/airflow/traits/task.rs delete mode 100644 src/airflow/traits/taskinstance.rs diff --git a/Cargo.lock b/Cargo.lock index 588f5726..e4f668ed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1286,7 +1286,6 @@ version = "0.13.2" dependencies = [ "ansi-to-tui", "anyhow", - "async-trait", "catppuccin", "chrono", "clap", diff --git a/Cargo.toml b/Cargo.toml index 48570292..09a37a66 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -77,7 +77,6 @@ formula = "flowrs" ansi-to-tui = { version = "8.0.1" } anyhow = { workspace = true } catppuccin = { workspace = true } -async-trait = { workspace = true } flowrs-config = { path = "crates/flowrs-config", version = "0.12.1" } flowrs-airflow = { path = "crates/flowrs-airflow", version = "0.11.1" } chrono = { workspace = true } diff --git a/src/airflow/client/impls/dag_ops.rs b/src/airflow/client/impls/dag_ops.rs index 60fe2af7..1daad51b 100644 --- a/src/airflow/client/impls/dag_ops.rs +++ b/src/airflow/client/impls/dag_ops.rs @@ -1,15 +1,12 @@ use anyhow::Result; -use async_trait::async_trait; use crate::airflow::client::convert_v1::v1_dag_collection_to_dag_list; use crate::airflow::client::convert_v2::v2_dag_list_to_dag_list; use crate::airflow::client::FlowrsClient; use crate::airflow::model::common::{Dag, DagList}; -use crate::airflow::traits::DagOperations; -#[async_trait] -impl DagOperations for FlowrsClient { - async fn list_dags(&self) -> Result { +impl FlowrsClient { + pub async fn list_dags(&self) -> Result { match self { Self::V1(client) => { let response = client.fetch_dags().await?; @@ -22,21 +19,21 @@ impl DagOperations for FlowrsClient { } } - async fn toggle_dag(&self, dag_id: &str, is_paused: bool) -> Result<()> { + pub async fn toggle_dag(&self, dag_id: &str, is_paused: bool) -> Result<()> { match self { Self::V1(client) => client.patch_dag_pause(dag_id, is_paused).await, Self::V2(client) => client.patch_dag_pause(dag_id, is_paused).await, } } - async fn get_dag_code(&self, dag: &Dag) -> Result { + pub async fn get_dag_code(&self, dag: &Dag) -> Result { match self { Self::V1(client) => client.fetch_dag_code(&dag.file_token).await, Self::V2(client) => client.fetch_dag_code(&dag.dag_id).await, } } - async fn get_dag_params(&self, dag_id: &str) -> Result> { + pub async fn get_dag_params(&self, dag_id: &str) -> Result> { match self { Self::V1(client) => client.fetch_dag_params(dag_id).await, Self::V2(client) => client.fetch_dag_params(dag_id).await, diff --git a/src/airflow/client/impls/dagrun_ops.rs b/src/airflow/client/impls/dagrun_ops.rs index 1dd30a0c..0369dcad 100644 --- a/src/airflow/client/impls/dagrun_ops.rs +++ b/src/airflow/client/impls/dagrun_ops.rs @@ -1,15 +1,12 @@ use anyhow::Result; -use async_trait::async_trait; use crate::airflow::client::convert_v1::v1_dagrun_collection_to_list; use crate::airflow::client::convert_v2::v2_dagrun_list_to_list; use crate::airflow::client::FlowrsClient; use crate::airflow::model::common::DagRunList; -use crate::airflow::traits::DagRunOperations; -#[async_trait] -impl DagRunOperations for FlowrsClient { - async fn list_dagruns(&self, dag_id: &str) -> Result { +impl FlowrsClient { + pub async fn list_dagruns(&self, dag_id: &str) -> Result { match self { Self::V1(client) => { let response = client.fetch_dagruns(dag_id).await?; @@ -22,21 +19,21 @@ impl DagRunOperations for FlowrsClient { } } - async fn mark_dag_run(&self, dag_id: &str, dag_run_id: &str, status: &str) -> Result<()> { + pub async fn mark_dag_run(&self, dag_id: &str, dag_run_id: &str, status: &str) -> Result<()> { match self { Self::V1(client) => client.patch_dag_run(dag_id, dag_run_id, status).await, Self::V2(client) => client.patch_dag_run(dag_id, dag_run_id, status).await, } } - async fn clear_dagrun(&self, dag_id: &str, dag_run_id: &str) -> Result<()> { + pub async fn clear_dagrun(&self, dag_id: &str, dag_run_id: &str) -> Result<()> { match self { Self::V1(client) => client.post_clear_dagrun(dag_id, dag_run_id).await, Self::V2(client) => client.post_clear_dagrun(dag_id, dag_run_id).await, } } - async fn trigger_dag_run( + pub async fn trigger_dag_run( &self, dag_id: &str, logical_date: Option<&str>, diff --git a/src/airflow/client/impls/dagstats_ops.rs b/src/airflow/client/impls/dagstats_ops.rs index f9be9723..4ee02fdc 100644 --- a/src/airflow/client/impls/dagstats_ops.rs +++ b/src/airflow/client/impls/dagstats_ops.rs @@ -1,15 +1,12 @@ use anyhow::Result; -use async_trait::async_trait; use crate::airflow::client::convert_v1::v1_dagstats_to_response; use crate::airflow::client::convert_v2::v2_dagstats_to_response; use crate::airflow::client::FlowrsClient; use crate::airflow::model::common::DagStatsResponse; -use crate::airflow::traits::DagStatsOperations; -#[async_trait] -impl DagStatsOperations for FlowrsClient { - async fn get_dag_stats(&self, dag_ids: Vec<&str>) -> Result { +impl FlowrsClient { + pub async fn get_dag_stats(&self, dag_ids: Vec<&str>) -> Result { match self { Self::V1(client) => { let response = client.fetch_dag_stats(dag_ids).await?; diff --git a/src/airflow/client/impls/log_ops.rs b/src/airflow/client/impls/log_ops.rs index c20a4236..88d6ff3b 100644 --- a/src/airflow/client/impls/log_ops.rs +++ b/src/airflow/client/impls/log_ops.rs @@ -1,15 +1,12 @@ use anyhow::Result; -use async_trait::async_trait; use crate::airflow::client::convert_v1::v1_log_to_log; use crate::airflow::client::convert_v2::v2_log_to_log; use crate::airflow::client::FlowrsClient; use crate::airflow::model::common::Log; -use crate::airflow::traits::LogOperations; -#[async_trait] -impl LogOperations for FlowrsClient { - async fn get_task_logs( +impl FlowrsClient { + pub async fn get_task_logs( &self, dag_id: &str, dag_run_id: &str, diff --git a/src/airflow/client/impls/task_ops.rs b/src/airflow/client/impls/task_ops.rs index 9132f63b..add542d4 100644 --- a/src/airflow/client/impls/task_ops.rs +++ b/src/airflow/client/impls/task_ops.rs @@ -1,15 +1,12 @@ use anyhow::Result; -use async_trait::async_trait; use crate::airflow::client::convert_v1::v1_task_collection_to_list; use crate::airflow::client::convert_v2::v2_task_collection_to_list; use crate::airflow::client::FlowrsClient; use crate::airflow::model::common::TaskList; -use crate::airflow::traits::TaskOperations; -#[async_trait] -impl TaskOperations for FlowrsClient { - async fn list_tasks(&self, dag_id: &str) -> Result { +impl FlowrsClient { + pub async fn list_tasks(&self, dag_id: &str) -> Result { match self { Self::V1(client) => { let response = client.fetch_tasks(dag_id).await?; diff --git a/src/airflow/client/impls/taskinstance_ops.rs b/src/airflow/client/impls/taskinstance_ops.rs index d563fa51..9b7284e1 100644 --- a/src/airflow/client/impls/taskinstance_ops.rs +++ b/src/airflow/client/impls/taskinstance_ops.rs @@ -1,5 +1,4 @@ use anyhow::Result; -use async_trait::async_trait; use crate::airflow::client::convert_v1::{ v1_task_instance_collection_to_list, v1_task_instance_try_to_gantt, @@ -9,11 +8,9 @@ use crate::airflow::client::convert_v2::{ }; use crate::airflow::client::FlowrsClient; use crate::airflow::model::common::{TaskInstanceList, TaskTryGantt}; -use crate::airflow::traits::TaskInstanceOperations; -#[async_trait] -impl TaskInstanceOperations for FlowrsClient { - async fn list_task_instances( +impl FlowrsClient { + pub async fn list_task_instances( &self, dag_id: &str, dag_run_id: &str, @@ -30,7 +27,7 @@ impl TaskInstanceOperations for FlowrsClient { } } - async fn list_task_instance_tries( + pub async fn list_task_instance_tries( &self, dag_id: &str, dag_run_id: &str, @@ -60,7 +57,7 @@ impl TaskInstanceOperations for FlowrsClient { } } - async fn mark_task_instance( + pub async fn mark_task_instance( &self, dag_id: &str, dag_run_id: &str, @@ -81,7 +78,7 @@ impl TaskInstanceOperations for FlowrsClient { } } - async fn clear_task_instance( + pub async fn clear_task_instance( &self, dag_id: &str, dag_run_id: &str, diff --git a/src/airflow/client/mod.rs b/src/airflow/client/mod.rs index 2bafaa5c..545caf9f 100644 --- a/src/airflow/client/mod.rs +++ b/src/airflow/client/mod.rs @@ -9,11 +9,10 @@ use flowrs_airflow::client::{BaseClient, V1Client, V2Client}; use flowrs_airflow::{AirflowConfig, AirflowVersion}; use crate::airflow::model::common::OpenItem; -use crate::airflow::traits::AirflowClient; use open_url::{build_v1_open_url, build_v2_open_url}; -/// Wrapper enum that owns a versioned Airflow HTTP client and implements the TUI trait layer. +/// Wrapper enum that owns a versioned Airflow HTTP client and exposes the TUI-facing operations. #[derive(Debug)] pub enum FlowrsClient { V1(V1Client), @@ -31,8 +30,8 @@ impl FlowrsClient { } } -impl AirflowClient for FlowrsClient { - fn build_open_url(&self, item: &OpenItem) -> Result { +impl FlowrsClient { + pub fn build_open_url(&self, item: &OpenItem) -> Result { match self { Self::V1(client) => build_v1_open_url(client.endpoint(), item), Self::V2(client) => build_v2_open_url(client.endpoint(), item), diff --git a/src/airflow/mod.rs b/src/airflow/mod.rs index faec7e10..80121bcd 100644 --- a/src/airflow/mod.rs +++ b/src/airflow/mod.rs @@ -1,4 +1,3 @@ pub mod client; pub mod graph; pub mod model; -pub mod traits; diff --git a/src/airflow/traits/dag.rs b/src/airflow/traits/dag.rs deleted file mode 100644 index 5ec82941..00000000 --- a/src/airflow/traits/dag.rs +++ /dev/null @@ -1,20 +0,0 @@ -use anyhow::Result; -use async_trait::async_trait; - -use crate::airflow::model::common::{Dag, DagList}; - -/// Trait for DAG operations -#[async_trait] -pub trait DagOperations: Send + Sync { - /// List all DAGs - async fn list_dags(&self) -> Result; - - /// Toggle a DAG's paused state - async fn toggle_dag(&self, dag_id: &str, is_paused: bool) -> Result<()>; - - /// Get DAG source code (uses `file_token` in v1, `dag_id` in v2) - async fn get_dag_code(&self, dag: &Dag) -> Result; - - /// Get DAG params schema (for trigger popup) - async fn get_dag_params(&self, dag_id: &str) -> Result>; -} diff --git a/src/airflow/traits/dagrun.rs b/src/airflow/traits/dagrun.rs deleted file mode 100644 index a9ba9dac..00000000 --- a/src/airflow/traits/dagrun.rs +++ /dev/null @@ -1,25 +0,0 @@ -use anyhow::Result; -use async_trait::async_trait; - -use crate::airflow::model::common::DagRunList; - -/// Trait for DAG Run operations -#[async_trait] -pub trait DagRunOperations: Send + Sync { - /// List DAG runs for a specific DAG - async fn list_dagruns(&self, dag_id: &str) -> Result; - - /// Mark a DAG run with a specific status - async fn mark_dag_run(&self, dag_id: &str, dag_run_id: &str, status: &str) -> Result<()>; - - /// Clear a DAG run - async fn clear_dagrun(&self, dag_id: &str, dag_run_id: &str) -> Result<()>; - - /// Trigger a new DAG run - async fn trigger_dag_run( - &self, - dag_id: &str, - logical_date: Option<&str>, - conf: Option, - ) -> Result<()>; -} diff --git a/src/airflow/traits/dagstats.rs b/src/airflow/traits/dagstats.rs deleted file mode 100644 index 2f5c8706..00000000 --- a/src/airflow/traits/dagstats.rs +++ /dev/null @@ -1,11 +0,0 @@ -use anyhow::Result; -use async_trait::async_trait; - -use crate::airflow::model::common::DagStatsResponse; - -/// Trait for DAG Statistics operations -#[async_trait] -pub trait DagStatsOperations: Send + Sync { - /// Get DAG statistics for the specified DAG IDs - async fn get_dag_stats(&self, dag_ids: Vec<&str>) -> Result; -} diff --git a/src/airflow/traits/log.rs b/src/airflow/traits/log.rs deleted file mode 100644 index ac10f78e..00000000 --- a/src/airflow/traits/log.rs +++ /dev/null @@ -1,17 +0,0 @@ -use anyhow::Result; -use async_trait::async_trait; - -use crate::airflow::model::common::Log; - -/// Trait for Log operations -#[async_trait] -pub trait LogOperations: Send + Sync { - /// Get task logs for a specific task instance and try number - async fn get_task_logs( - &self, - dag_id: &str, - dag_run_id: &str, - task_id: &str, - task_try: u32, - ) -> Result; -} diff --git a/src/airflow/traits/mod.rs b/src/airflow/traits/mod.rs deleted file mode 100644 index edb634b5..00000000 --- a/src/airflow/traits/mod.rs +++ /dev/null @@ -1,37 +0,0 @@ -pub mod dag; -pub mod dagrun; -pub mod dagstats; -pub mod log; -pub mod task; -pub mod taskinstance; - -pub use dag::DagOperations; -pub use dagrun::DagRunOperations; -pub use dagstats::DagStatsOperations; -pub use log::LogOperations; -pub use task::TaskOperations; -pub use taskinstance::TaskInstanceOperations; - -use super::model::common::OpenItem; -use anyhow::Result; - -/// Super-trait combining all Airflow API operations. -/// -/// This trait can be implemented by different API versions (v1 for Airflow v2, v2 for Airflow v3) -/// to provide a consistent interface for interacting with Airflow. -pub trait AirflowClient: - std::fmt::Debug - + DagOperations - + DagRunOperations - + TaskInstanceOperations - + LogOperations - + DagStatsOperations - + TaskOperations -{ - /// Build the appropriate web UI URL for opening an item in the browser. - /// The URL structure differs between Airflow v2 and v3. - /// - /// # Errors - /// Returns an error if the URL cannot be constructed for the given item. - fn build_open_url(&self, item: &OpenItem) -> Result; -} diff --git a/src/airflow/traits/task.rs b/src/airflow/traits/task.rs deleted file mode 100644 index b125cf49..00000000 --- a/src/airflow/traits/task.rs +++ /dev/null @@ -1,11 +0,0 @@ -use anyhow::Result; -use async_trait::async_trait; - -use crate::airflow::model::common::TaskList; - -/// Trait for task definition operations -#[async_trait] -pub trait TaskOperations: Send + Sync { - /// List all tasks for a DAG - async fn list_tasks(&self, dag_id: &str) -> Result; -} diff --git a/src/airflow/traits/taskinstance.rs b/src/airflow/traits/taskinstance.rs deleted file mode 100644 index d44e5335..00000000 --- a/src/airflow/traits/taskinstance.rs +++ /dev/null @@ -1,37 +0,0 @@ -use anyhow::Result; -use async_trait::async_trait; - -use crate::airflow::model::common::{TaskInstanceList, TaskTryGantt}; - -/// Trait for Task Instance operations -#[async_trait] -pub trait TaskInstanceOperations: Send + Sync { - /// List task instances for a specific DAG run - async fn list_task_instances(&self, dag_id: &str, dag_run_id: &str) - -> Result; - - /// List all tries for a specific task instance (for Gantt chart retry visualization) - async fn list_task_instance_tries( - &self, - dag_id: &str, - dag_run_id: &str, - task_id: &str, - ) -> Result>; - - /// Mark a task instance with a specific status - async fn mark_task_instance( - &self, - dag_id: &str, - dag_run_id: &str, - task_id: &str, - status: &str, - ) -> Result<()>; - - /// Clear a task instance - async fn clear_task_instance( - &self, - dag_id: &str, - dag_run_id: &str, - task_id: &str, - ) -> Result<()>; -} diff --git a/src/app/state/environment_state.rs b/src/app/state/environment_state.rs index ec247d2b..fc148bdb 100644 --- a/src/app/state/environment_state.rs +++ b/src/app/state/environment_state.rs @@ -1,10 +1,10 @@ +use crate::airflow::client::FlowrsClient; use std::collections::HashMap; use std::sync::Arc; use crate::airflow::model::common::{ Dag, DagId, DagRun, DagRunId, DagStatistic, EnvironmentKey, Log, TaskId, TaskInstance, }; -use crate::airflow::traits::AirflowClient as AirflowClientTrait; /// Flat, request-keyed cache for a single Airflow environment. /// @@ -14,7 +14,7 @@ use crate::airflow::traits::AirflowClient as AirflowClientTrait; /// allocations. #[derive(Debug, Clone)] pub struct EnvironmentData { - pub client: Arc, + pub client: Arc, /// Result of `list_dags()` — sorted alphabetically by `dag_id` on write. pub dags: Vec, @@ -38,7 +38,7 @@ pub struct EnvironmentData { } impl EnvironmentData { - pub fn new(client: Arc) -> Self { + pub fn new(client: Arc) -> Self { Self { client, dags: Vec::new(), diff --git a/src/app/worker/browser.rs b/src/app/worker/browser.rs index 15d7f226..9d0255ca 100644 --- a/src/app/worker/browser.rs +++ b/src/app/worker/browser.rs @@ -1,9 +1,9 @@ +use crate::airflow::client::FlowrsClient; use std::sync::{Arc, Mutex}; use anyhow::{Context, Result}; use crate::airflow::model::common::OpenItem; -use crate::airflow::traits::AirflowClient; use crate::app::state::App; /// Open an item (DAG, DAG run, task instance, etc.) in the browser. @@ -11,17 +11,13 @@ use crate::app::state::App; /// Any failure (no active server, an unbuildable URL, or the browser refusing /// to launch) is surfaced to the user via the active panel's error popup /// instead of being silently logged. -pub fn handle_open_item(app: &Arc>, client: &Arc, item: OpenItem) { +pub fn handle_open_item(app: &Arc>, client: &Arc, item: OpenItem) { if let Err(e) = try_open_item(app, client, item) { app.lock().unwrap().show_error(vec![e.to_string()]); } } -fn try_open_item( - app: &Arc>, - client: &Arc, - item: OpenItem, -) -> Result<()> { +fn try_open_item(app: &Arc>, client: &Arc, item: OpenItem) -> Result<()> { // For Config items, look up the endpoint from active_server instead of using the passed string let final_item = if let OpenItem::Config(_) = &item { let app_lock = app.lock().unwrap(); diff --git a/src/app/worker/dagruns.rs b/src/app/worker/dagruns.rs index d4164dbe..1e01f6eb 100644 --- a/src/app/worker/dagruns.rs +++ b/src/app/worker/dagruns.rs @@ -1,9 +1,9 @@ +use crate::airflow::client::FlowrsClient; use std::sync::{Arc, Mutex}; use log::debug; use crate::airflow::model::common::{DagId, DagRunId, DagRunState}; -use crate::airflow::traits::AirflowClient; use crate::app::model::dagruns::popup::mark::MarkState; use crate::app::state::App; @@ -13,7 +13,7 @@ use crate::app::state::App; /// results are written to the correct environment even if the active one changes. pub async fn handle_update_dag_runs( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, env_name: &str, ) { @@ -40,7 +40,7 @@ pub async fn handle_update_dag_runs( /// Handle clearing a DAG run (resets all task instances). pub async fn handle_clear_dag_run( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, dag_run_id: &DagRunId, ) { @@ -56,7 +56,7 @@ pub async fn handle_clear_dag_run( /// Handle marking a DAG run with a new state (success/failed). pub async fn handle_mark_dag_run( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, dag_run_id: &DagRunId, status: MarkState, @@ -81,7 +81,7 @@ pub async fn handle_mark_dag_run( /// Handle triggering a new DAG run. pub async fn handle_trigger_dag_run( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, env_name: &str, conf: Option, diff --git a/src/app/worker/dags.rs b/src/app/worker/dags.rs index 36edf6ce..35c4eecd 100644 --- a/src/app/worker/dags.rs +++ b/src/app/worker/dags.rs @@ -1,9 +1,9 @@ +use crate::airflow::client::FlowrsClient; use std::sync::{Arc, Mutex}; use log::warn; use crate::airflow::model::common::DagId; -use crate::airflow::traits::AirflowClient; use crate::app::model::dagruns::popup::trigger::TriggerDagRunPopUp; use crate::app::model::dagruns::popup::DagRunPopUp; use crate::app::model::dagruns::DagCodeView; @@ -19,7 +19,7 @@ use crate::app::state::{App, Panel}; /// results are written to the correct environment even if the active one changes. pub async fn handle_update_dags_and_stats( app: &Arc>, - client: &Arc, + client: &Arc, env_name: &str, ) { // Snapshot cached DAG IDs from the originating environment for the stats request @@ -121,7 +121,7 @@ pub async fn handle_update_dags_and_stats( /// Handle toggling the paused state of a DAG. pub async fn handle_toggle_dag( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, is_paused: bool, ) { @@ -135,7 +135,7 @@ pub async fn handle_toggle_dag( /// Handle fetching the DAG source code. pub async fn handle_get_dag_code( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, ) { let current_dag = { @@ -179,7 +179,7 @@ pub async fn handle_get_dag_code( /// fallback when the fetch fails. pub async fn handle_get_dag_params( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, env_name: &str, ) { diff --git a/src/app/worker/logs.rs b/src/app/worker/logs.rs index fa23cf87..5c1fa0fd 100644 --- a/src/app/worker/logs.rs +++ b/src/app/worker/logs.rs @@ -1,10 +1,10 @@ +use crate::airflow::client::FlowrsClient; use std::sync::{Arc, Mutex}; use futures::future::join_all; use log::debug; use crate::airflow::model::common::{DagId, DagRunId, TaskId}; -use crate::airflow::traits::AirflowClient; use crate::app::model::popup::error::ErrorPopup; use crate::app::state::App; @@ -14,7 +14,7 @@ use crate::app::state::App; /// results are written to the correct environment even if the active one changes. pub async fn handle_update_task_logs( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, dag_run_id: &DagRunId, task_id: &TaskId, diff --git a/src/app/worker/taskinstances.rs b/src/app/worker/taskinstances.rs index 449bc5a1..cfa2e81f 100644 --- a/src/app/worker/taskinstances.rs +++ b/src/app/worker/taskinstances.rs @@ -1,3 +1,4 @@ +use crate::airflow::client::FlowrsClient; use std::collections::HashSet; use std::sync::{Arc, Mutex}; @@ -5,7 +6,6 @@ use futures::future::join_all; use log::debug; use crate::airflow::model::common::{DagId, DagRunId, GanttData, TaskId, TaskInstanceState}; -use crate::airflow::traits::AirflowClient; use crate::app::model::taskinstances::popup::mark::MarkState; use crate::app::state::App; @@ -19,7 +19,7 @@ use crate::app::state::App; /// atomically under a single lock. pub async fn handle_update_task_instances( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, dag_run_id: &DagRunId, env_name: &str, @@ -77,7 +77,7 @@ pub async fn handle_update_task_instances( /// Handle clearing a task instance (resets it to be re-run). pub async fn handle_clear_task_instance( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, dag_run_id: &DagRunId, task_id: &TaskId, @@ -96,7 +96,7 @@ pub async fn handle_clear_task_instance( /// Handle marking a task instance with a new state (success/failed). pub async fn handle_mark_task_instance( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &DagId, dag_run_id: &DagRunId, task_id: &TaskId, diff --git a/src/app/worker/tasks.rs b/src/app/worker/tasks.rs index bfa796ba..a8497d47 100644 --- a/src/app/worker/tasks.rs +++ b/src/app/worker/tasks.rs @@ -1,19 +1,15 @@ +use crate::airflow::client::FlowrsClient; use std::sync::{Arc, Mutex}; use log::debug; use crate::airflow::graph::TaskGraph; -use crate::airflow::traits::AirflowClient; use crate::app::model::dagruns::popup::DagRunPopUp; use crate::app::model::taskinstances::popup::graph::DagGraphPopup; use crate::app::state::App; /// Handle fetching task definitions and building the task graph -pub async fn handle_update_tasks( - app: &Arc>, - client: &Arc, - dag_id: &str, -) { +pub async fn handle_update_tasks(app: &Arc>, client: &Arc, dag_id: &str) { debug!("Fetching tasks for DAG: {dag_id}"); match client.list_tasks(dag_id).await { @@ -40,7 +36,7 @@ pub async fn handle_update_tasks( /// then builds the graph popup and displays it on the dagrun panel. pub async fn handle_show_dag_graph( app: &Arc>, - client: &Arc, + client: &Arc, dag_id: &str, dag_run_id: &str, ) { diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 77304e33..446152f2 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -9,7 +9,6 @@ use std::sync::Arc; use flowrs_config::{AirflowAuth, AirflowConfig, AirflowVersion, BasicAuth, TokenSource}; use flowrs_tui::airflow::client::FlowrsClient; -use flowrs_tui::airflow::traits::AirflowClient; /// Check if we should run tests for a specific API version. /// Returns false if `TEST_AIRFLOW_URL` is not set (required for all API tests). @@ -54,7 +53,7 @@ async fn get_jwt_token(url: &str, username: &str, password: &str) -> anyhow::Res } /// Create a test client from environment variables -pub fn create_test_client() -> anyhow::Result> { +pub fn create_test_client() -> anyhow::Result> { let url = env::var("TEST_AIRFLOW_URL").expect("TEST_AIRFLOW_URL must be set"); let username = env::var("TEST_AIRFLOW_USERNAME").unwrap_or_else(|_| "airflow".to_string()); let password = env::var("TEST_AIRFLOW_PASSWORD").unwrap_or_else(|_| "airflow".to_string()); @@ -80,7 +79,7 @@ pub fn create_test_client() -> anyhow::Result> { } /// Create a test client for Airflow 3.x using JWT authentication -pub async fn create_test_client_v3() -> anyhow::Result> { +pub async fn create_test_client_v3() -> anyhow::Result> { let url = env::var("TEST_AIRFLOW_URL").expect("TEST_AIRFLOW_URL must be set"); let username = env::var("TEST_AIRFLOW_USERNAME").unwrap_or_else(|_| "airflow".to_string()); let password = env::var("TEST_AIRFLOW_PASSWORD").unwrap_or_else(|_| "airflow".to_string());