diff --git a/crates/flowrs-airflow/src/client/v1/dagrun.rs b/crates/flowrs-airflow/src/client/v1/dagrun.rs index 86bde95c..0286ede5 100644 --- a/crates/flowrs-airflow/src/client/v1/dagrun.rs +++ b/crates/flowrs-airflow/src/client/v1/dagrun.rs @@ -22,18 +22,6 @@ impl V1Client { Ok(dagruns) } - pub async fn fetch_all_dagruns(&self) -> Result { - let response: Response = self - .base_api(Method::POST, "dags/~/dagRuns/list") - .await? - .json(&serde_json::json!({"page_limit": 200})) - .send() - .await? - .error_for_status()?; - let dagruns: model::dagrun::DAGRunCollectionResponse = response.json().await?; - Ok(dagruns) - } - pub async fn patch_dag_run(&self, dag_id: &str, dag_run_id: &str, status: &str) -> Result<()> { self.base_api( Method::PATCH, diff --git a/crates/flowrs-airflow/src/client/v1/taskinstance.rs b/crates/flowrs-airflow/src/client/v1/taskinstance.rs index 9efcbc0d..27af7795 100644 --- a/crates/flowrs-airflow/src/client/v1/taskinstance.rs +++ b/crates/flowrs-airflow/src/client/v1/taskinstance.rs @@ -66,53 +66,6 @@ impl V1Client { }) } - pub async fn fetch_all_task_instances( - &self, - ) -> Result { - let mut all_task_instances = Vec::new(); - let mut offset = 0; - let limit = 100; - let mut total_entries; - - loop { - let response: Response = self - .base_api(Method::GET, "dags/~/dagRuns/~/taskInstances") - .await? - .query(&[("limit", limit.to_string()), ("offset", offset.to_string())]) - .send() - .await? - .error_for_status()?; - - let response_text = response.text().await?; - let page: model::taskinstance::TaskInstanceCollectionResponse = - parse_json_response(&response_text, "all task instances response")?; - - total_entries = page.total_entries; - let fetched_count = page.task_instances.len(); - all_task_instances.extend(page.task_instances); - - debug!("Fetched {fetched_count} task instances (all), offset: {offset}, total: {total_entries}"); - - let total_usize = usize::try_from(total_entries).unwrap_or(usize::MAX); - if fetched_count < limit || all_task_instances.len() >= total_usize { - break; - } - - offset += fetched_count; - } - - info!( - "Fetched total {} task instances (all) out of {}", - all_task_instances.len(), - total_entries - ); - - Ok(model::taskinstance::TaskInstanceCollectionResponse { - task_instances: all_task_instances, - total_entries, - }) - } - pub async fn fetch_task_instance_tries( &self, dag_id: &str, diff --git a/crates/flowrs-airflow/src/client/v2/dagrun.rs b/crates/flowrs-airflow/src/client/v2/dagrun.rs index 461aebb5..714470a5 100644 --- a/crates/flowrs-airflow/src/client/v2/dagrun.rs +++ b/crates/flowrs-airflow/src/client/v2/dagrun.rs @@ -18,18 +18,6 @@ impl V2Client { Ok(dagruns) } - pub async fn fetch_all_dagruns(&self) -> Result { - let response: Response = self - .base_api(Method::POST, "dags/~/dagRuns/list") - .await? - .json(&serde_json::json!({"page_limit": 200})) - .send() - .await? - .error_for_status()?; - let dagruns: model::dagrun::DagRunList = response.json().await?; - Ok(dagruns) - } - pub async fn patch_dag_run(&self, dag_id: &str, dag_run_id: &str, status: &str) -> Result<()> { self.base_api( Method::PATCH, diff --git a/crates/flowrs-airflow/src/client/v2/taskinstance.rs b/crates/flowrs-airflow/src/client/v2/taskinstance.rs index 49583004..dd81d4e1 100644 --- a/crates/flowrs-airflow/src/client/v2/taskinstance.rs +++ b/crates/flowrs-airflow/src/client/v2/taskinstance.rs @@ -60,55 +60,6 @@ impl V2Client { }) } - pub async fn fetch_all_task_instances(&self) -> Result { - let mut all_task_instances = Vec::new(); - let mut offset = 0; - let limit = 100; - let mut total_entries; - - loop { - let response: Response = self - .base_api(Method::GET, "dags/~/dagRuns/~/taskInstances") - .await? - .query(&[("limit", limit.to_string()), ("offset", offset.to_string())]) - .send() - .await? - .error_for_status()?; - - let page: model::taskinstance::TaskInstanceList = response.json().await?; - - total_entries = page.total_entries; - let fetched_count = page.task_instances.len(); - all_task_instances.extend(page.task_instances); - - debug!( - "Fetched {fetched_count} task instances (all), offset: {offset}, total: {total_entries}" - ); - - #[expect( - clippy::cast_possible_truncation, - clippy::cast_sign_loss, - reason = "duration/count values from the API are small and non-negative in practice" - )] - if fetched_count < limit || all_task_instances.len() >= total_entries as usize { - break; - } - - offset += limit; - } - - info!( - "Fetched total {} task instances (all) out of {}", - all_task_instances.len(), - total_entries - ); - - Ok(model::taskinstance::TaskInstanceList { - task_instances: all_task_instances, - total_entries, - }) - } - pub async fn fetch_task_instance_tries( &self, dag_id: &str, diff --git a/crates/flowrs-config/src/paths.rs b/crates/flowrs-config/src/paths.rs index a6c3b92d..2d3986d8 100644 --- a/crates/flowrs-config/src/paths.rs +++ b/crates/flowrs-config/src/paths.rs @@ -66,14 +66,6 @@ impl ConfigPaths { fn legacy_config_path() -> PathBuf { home_dir().unwrap_or_default().join(".flowrs") } - - /// Returns the XDG config directory (for creating if needed). - pub fn xdg_config_dir(&self) -> PathBuf { - self.write_path - .parent() - .expect("Config write path should have a parent directory") - .to_path_buf() - } } #[cfg(test)] diff --git a/src/airflow/client/impls/dagrun_ops.rs b/src/airflow/client/impls/dagrun_ops.rs index fec7706d..1dd30a0c 100644 --- a/src/airflow/client/impls/dagrun_ops.rs +++ b/src/airflow/client/impls/dagrun_ops.rs @@ -22,19 +22,6 @@ impl DagRunOperations for FlowrsClient { } } - async fn list_all_dagruns(&self) -> Result { - match self { - Self::V1(client) => { - let response = client.fetch_all_dagruns().await?; - Ok(v1_dagrun_collection_to_list(response)) - } - Self::V2(client) => { - let response = client.fetch_all_dagruns().await?; - Ok(v2_dagrun_list_to_list(response)) - } - } - } - 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, diff --git a/src/airflow/client/impls/taskinstance_ops.rs b/src/airflow/client/impls/taskinstance_ops.rs index 18d74185..d563fa51 100644 --- a/src/airflow/client/impls/taskinstance_ops.rs +++ b/src/airflow/client/impls/taskinstance_ops.rs @@ -30,19 +30,6 @@ impl TaskInstanceOperations for FlowrsClient { } } - async fn list_all_taskinstances(&self) -> Result { - match self { - Self::V1(client) => { - let response = client.fetch_all_task_instances().await?; - Ok(v1_task_instance_collection_to_list(response)) - } - Self::V2(client) => { - let response = client.fetch_all_task_instances().await?; - Ok(v2_task_instance_list_to_list(response)) - } - } - } - async fn list_task_instance_tries( &self, dag_id: &str, diff --git a/src/airflow/client/mod.rs b/src/airflow/client/mod.rs index 84fe40cc..2bafaa5c 100644 --- a/src/airflow/client/mod.rs +++ b/src/airflow/client/mod.rs @@ -32,13 +32,6 @@ impl FlowrsClient { } impl AirflowClient for FlowrsClient { - fn get_version(&self) -> AirflowVersion { - match self { - Self::V1(_) => AirflowVersion::V2, - Self::V2(_) => AirflowVersion::V3, - } - } - fn build_open_url(&self, item: &OpenItem) -> Result { match self { Self::V1(client) => build_v1_open_url(client.endpoint(), item), diff --git a/src/airflow/traits/dagrun.rs b/src/airflow/traits/dagrun.rs index 9c34cd8c..a9ba9dac 100644 --- a/src/airflow/traits/dagrun.rs +++ b/src/airflow/traits/dagrun.rs @@ -9,10 +9,6 @@ pub trait DagRunOperations: Send + Sync { /// List DAG runs for a specific DAG async fn list_dagruns(&self, dag_id: &str) -> Result; - /// List all DAG runs across all DAGs - #[allow(unused, reason = "trait method kept for API completeness")] - async fn list_all_dagruns(&self) -> 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<()>; diff --git a/src/airflow/traits/mod.rs b/src/airflow/traits/mod.rs index 02555c4f..edb634b5 100644 --- a/src/airflow/traits/mod.rs +++ b/src/airflow/traits/mod.rs @@ -14,7 +14,6 @@ pub use taskinstance::TaskInstanceOperations; use super::model::common::OpenItem; use anyhow::Result; -use flowrs_airflow::AirflowVersion; /// Super-trait combining all Airflow API operations. /// @@ -29,10 +28,6 @@ pub trait AirflowClient: + DagStatsOperations + TaskOperations { - /// Get the Airflow version this client is configured for - #[allow(unused, reason = "trait method kept for API completeness")] - fn get_version(&self) -> AirflowVersion; - /// Build the appropriate web UI URL for opening an item in the browser. /// The URL structure differs between Airflow v2 and v3. /// diff --git a/src/airflow/traits/taskinstance.rs b/src/airflow/traits/taskinstance.rs index 831bc43e..d44e5335 100644 --- a/src/airflow/traits/taskinstance.rs +++ b/src/airflow/traits/taskinstance.rs @@ -10,10 +10,6 @@ pub trait TaskInstanceOperations: Send + Sync { async fn list_task_instances(&self, dag_id: &str, dag_run_id: &str) -> Result; - /// List all task instances across all DAG runs - #[allow(unused, reason = "trait method kept for API completeness")] - async fn list_all_taskinstances(&self) -> Result; - /// List all tries for a specific task instance (for Gantt chart retry visualization) async fn list_task_instance_tries( &self, diff --git a/src/app/model.rs b/src/app/model.rs index b8fe4b9e..6d15bf87 100644 --- a/src/app/model.rs +++ b/src/app/model.rs @@ -56,15 +56,6 @@ impl KeyResult { KeyResult::Ignored => (Some(event.clone()), vec![]), } } - - /// Create from a simple bool (true = consumed, false = ignored) - pub fn from_consumed(consumed: bool) -> Self { - if consumed { - KeyResult::Consumed - } else { - KeyResult::Ignored - } - } } pub trait Model {