refactor(airflow): give the library a canonical error type

Cargo.toml

@@ -18,6 +18,7 @@ reqwest = { version = "0.12.28", features = ["json", "rustls-tls", "rustls-tls-n
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.150"
 strum = { version = "0.28.0", features = ["derive"] }
+thiserror = "2.0.18"
 time = { version = "0.3.47", features = ["serde", "serde-human-readable", "parsing", "macros"] }
 tokio = { version = "1.51.1", features = ["rt-multi-thread", "sync", "macros", "time"] }
 toml = "1.1.2"

crates/flowrs-airflow/Cargo.toml

@@ -19,6 +19,7 @@ anyhow = { workspace = true }
 async-trait = { workspace = true }
 clap = { workspace = true }
 strum = { workspace = true }
+thiserror = { workspace = true }
 aws-config = { version = "1.8.18", optional = true }
 aws-sdk-mwaa = { version = "1.108.0", optional = true }
 dirs = { workspace = true, optional = true }

crates/flowrs-airflow/src/client.rs

@@ -3,28 +3,22 @@ pub mod base;
 pub mod v1;
 pub mod v2;
 
-use anyhow::Result;
+use reqwest::Response;
+use serde::de::DeserializeOwned;
 
-use crate::config::{AirflowConfig, AirflowVersion};
+use crate::error::{AirflowError, Result};
 
 pub use base::BaseClient;
 pub use v1::V1Client;
 pub use v2::V2Client;
 
-/// Enum wrapping the versioned API clients.
-/// V1 is for Airflow v2 (uses /api/v1), V2 is for Airflow v3 (uses /api/v2).
-#[derive(Debug)]
-pub enum AirflowApiClient {
-    V1(V1Client),
-    V2(V2Client),
-}
-
-/// Create an Airflow API client based on the configuration version
-pub fn create_api_client(config: &AirflowConfig) -> Result<AirflowApiClient> {
-    let base = BaseClient::new(config.clone())?;
-
-    match config.version {
-        AirflowVersion::V2 => Ok(AirflowApiClient::V1(V1Client::new(base))), // V2 uses API v1
-        AirflowVersion::V3 => Ok(AirflowApiClient::V2(V2Client::new(base))), // V3 uses API v2
-    }
+/// Deserialize a response body, keeping a snippet of it in the error on failure.
+///
+/// The body is read as text first rather than using `Response::json`, because some
+/// deployments return payloads that do not match the documented schema (older v2.x
+/// versions omit `dag_display_name`, for instance) and the raw body is what makes
+/// those failures diagnosable.
+pub(crate) async fn read_json<T: DeserializeOwned>(response: Response, context: &str) -> Result<T> {
+    let body = response.text().await?;
+    serde_json::from_str(&body).map_err(|e| AirflowError::decode(context, &body, e))
 }

crates/flowrs-airflow/src/client/auth/basic.rs

@@ -1,4 +1,4 @@
-use anyhow::Result;
+use crate::error::Result;
 use async_trait::async_trait;
 use log::info;
 use reqwest::RequestBuilder;

crates/flowrs-airflow/src/client/auth/command.rs

@@ -1,12 +1,13 @@
 use std::fmt;
 use std::time::{Duration, Instant};
 
-use anyhow::{Context, Result};
+use anyhow::Context;
 use async_trait::async_trait;
 use log::info;
 use reqwest::RequestBuilder;
 
 use super::AuthProvider;
+use crate::error::{AirflowError, Result};
 
 /// How long a fetched token is reused before the helper command is run again.
 ///
@@ -32,7 +33,10 @@ impl CommandTokenProvider {
     }
 
     /// Run the helper command and return its trimmed token output.
-    async fn fetch_token(&self) -> Result<String> {
+    ///
+    /// Uses `anyhow` internally for the layered context messages; the chain is
+    /// flattened into `AirflowError::Auth` at the trait boundary.
+    async fn fetch_token(&self) -> anyhow::Result<String> {
         let cmd = self.cmd.clone();
         let output = tokio::task::spawn_blocking(move || {
             std::process::Command::new("sh")
@@ -81,7 +85,10 @@ impl AuthProvider for CommandTokenProvider {
 
         if !fresh {
             info!("🔑 Token Auth (command): refreshing via {}", self.cmd);
-            let token = self.fetch_token().await?;
+            let token = self
+                .fetch_token()
+                .await
+                .map_err(|e| AirflowError::auth("token command", &e))?;
             *cached = Some((token, Instant::now()));
         }
 

crates/flowrs-airflow/src/client/auth/mod.rs

@@ -2,7 +2,7 @@ mod basic;
 mod command;
 mod static_token;
 
-use anyhow::Result;
+use crate::error::Result;
 
 pub use basic::BasicAuthProvider;
 pub use command::CommandTokenProvider;
@@ -46,33 +46,35 @@ pub fn create_auth_provider(auth: &AirflowAuth) -> Result<Box<dyn AuthProvider>>
         #[cfg(feature = "conveyor")]
         AirflowAuth::Conveyor => Ok(Box::new(ConveyorAuthProvider::new())),
         #[cfg(not(feature = "conveyor"))]
-        AirflowAuth::Conveyor => {
-            anyhow::bail!("Conveyor support not compiled. Enable the 'conveyor' feature.")
-        }
+        AirflowAuth::Conveyor => Err(crate::error::AirflowError::FeatureNotEnabled {
+            service: "Conveyor",
+            feature: "conveyor",
+        }),
         #[cfg(feature = "mwaa")]
         AirflowAuth::Mwaa(mwaa_auth) => Ok(Box::new(MwaaAuthProvider::from(mwaa_auth))),
         #[cfg(not(feature = "mwaa"))]
-        AirflowAuth::Mwaa(_) => {
-            anyhow::bail!("MWAA support not compiled. Enable the 'mwaa' feature.")
-        }
+        AirflowAuth::Mwaa(_) => Err(crate::error::AirflowError::FeatureNotEnabled {
+            service: "MWAA",
+            feature: "mwaa",
+        }),
         #[cfg(feature = "astronomer")]
         AirflowAuth::Astronomer(astro_auth) => {
             Ok(Box::new(AstronomerAuthProvider::from(astro_auth)))
         }
         #[cfg(not(feature = "astronomer"))]
-        AirflowAuth::Astronomer(_) => {
-            anyhow::bail!("Astronomer support not compiled. Enable the 'astronomer' feature.")
-        }
+        AirflowAuth::Astronomer(_) => Err(crate::error::AirflowError::FeatureNotEnabled {
+            service: "Astronomer",
+            feature: "astronomer",
+        }),
         #[cfg(feature = "composer")]
         AirflowAuth::Composer(composer_auth) => {
             Ok(Box::new(ComposerAuthProvider::new(composer_auth)?))
         }
         #[cfg(not(feature = "composer"))]
-        AirflowAuth::Composer(_) => {
-            anyhow::bail!(
-                "Google Cloud Composer support not compiled. Enable the 'composer' feature."
-            )
-        }
+        AirflowAuth::Composer(_) => Err(crate::error::AirflowError::FeatureNotEnabled {
+            service: "Google Cloud Composer",
+            feature: "composer",
+        }),
     }
 }
 

crates/flowrs-airflow/src/client/auth/static_token.rs

@@ -1,4 +1,4 @@
-use anyhow::Result;
+use crate::error::Result;
 use async_trait::async_trait;
 use log::info;
 use reqwest::RequestBuilder;

crates/flowrs-airflow/src/client/base.rs

@@ -1,18 +1,20 @@
-use anyhow::Result;
 use log::{debug, warn};
-use reqwest::{Method, Url};
-use std::convert::TryFrom;
+use reqwest::{Method, RequestBuilder, Response, Url};
 use std::fmt;
 use std::time::Duration;
 
 use super::auth::{create_auth_provider, AuthProvider};
 use crate::config::AirflowConfig;
+use crate::error::{AirflowError, Result};
 
 /// Base HTTP client for Airflow API communication.
 /// Handles authentication and provides base request building functionality.
 pub struct BaseClient {
-    pub client: reqwest::Client,
-    pub config: AirflowConfig,
+    client: reqwest::Client,
+    config: AirflowConfig,
+    /// The endpoint parsed once at construction, so an unusable URL is reported
+    /// when the client is created rather than on every request.
+    endpoint: Url,
     auth_provider: Box<dyn AuthProvider>,
 }
 
@@ -21,6 +23,7 @@ impl fmt::Debug for BaseClient {
         f.debug_struct("BaseClient")
             .field("client", &self.client)
             .field("config", &self.config)
+            .field("endpoint", &self.endpoint)
             .field("auth_provider", &"<AuthProvider>")
             .finish()
     }
@@ -34,6 +37,16 @@ impl BaseClient {
                 config.name
             );
         }
+        let mut endpoint = Url::parse(&config.endpoint)
+            .map_err(|e| AirflowError::invalid_url(config.endpoint.clone(), e))?;
+        // Ensure the base path ends with a slash so `base_api`'s relative `join`
+        // extends the endpoint rather than replacing its final segment. Without
+        // this, a reverse-proxy prefix such as `/airflow` would be dropped.
+        if !endpoint.path().ends_with('/') {
+            let with_slash = format!("{}/", endpoint.path());
+            endpoint.set_path(&with_slash);
+        }
+
         let client = reqwest::Client::builder()
             .timeout(Duration::from_secs(config.timeout_secs))
             .use_rustls_tls()
@@ -45,30 +58,106 @@ impl BaseClient {
         Ok(Self {
             client,
             config,
+            endpoint,
             auth_provider,
         })
     }
 
+    /// The base URL of the Airflow deployment this client talks to.
+    pub const fn endpoint(&self) -> &Url {
+        &self.endpoint
+    }
+
+    pub const fn config(&self) -> &AirflowConfig {
+        &self.config
+    }
+
     /// Build a base request with authentication for the specified API version
-    pub async fn base_api(
+    pub(crate) async fn base_api(
         &self,
         method: Method,
         endpoint: &str,
         api_version: &str,
-    ) -> Result<reqwest::RequestBuilder> {
-        let base_url = Url::parse(&self.config.endpoint)?;
-        let url = base_url.join(format!("{api_version}/{endpoint}").as_str())?;
+    ) -> Result<RequestBuilder> {
+        let path = format!("{api_version}/{endpoint}");
+        let url = self
+            .endpoint
+            .join(&path)
+            .map_err(|e| AirflowError::invalid_url(path, e))?;
         debug!("🔗 Request URL: {url}");
 
         let request = self.client.request(method, url);
         self.auth_provider.authenticate(request).await
     }
+
+    /// Send a request, turning a non-success status into [`AirflowError::Status`].
+    ///
+    /// The request is built rather than sent through `RequestBuilder::send` so that
+    /// the method and URL are available for the error message, and the body is read
+    /// instead of calling `error_for_status`, which discards Airflow's `detail`.
+    pub(crate) async fn execute(&self, request: RequestBuilder) -> Result<Response> {
+        let request = request.build()?;
+        let method = request.method().clone();
+        let url = request.url().clone();
+
+        let response = self.client.execute(request).await?;
+        let status = response.status();
+        if status.is_success() {
+            return Ok(response);
+        }
+
+        let body = response.text().await.unwrap_or_default();
+        Err(AirflowError::status(&method, &url, status, &body))
+    }
 }
 
 impl TryFrom<&AirflowConfig> for BaseClient {
-    type Error = anyhow::Error;
+    type Error = AirflowError;
 
-    fn try_from(config: &AirflowConfig) -> Result<Self, Self::Error> {
+    fn try_from(config: &AirflowConfig) -> Result<Self> {
         Self::new(config.clone())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::auth::{AirflowAuth, TokenSource};
+
+    fn config(endpoint: &str) -> AirflowConfig {
+        AirflowConfig {
+            name: "test".to_string(),
+            endpoint: endpoint.to_string(),
+            auth: AirflowAuth::Token(TokenSource::Static {
+                token: "tok".to_string(),
+            }),
+            managed: None,
+            version: crate::config::AirflowVersion::V3,
+            timeout_secs: crate::config::default_timeout(),
+            insecure: false,
+        }
+    }
+
+    #[test]
+    fn rejects_an_unusable_endpoint_at_construction() {
+        let error = BaseClient::new(config("not a url")).expect_err("should reject");
+        assert!(
+            matches!(error, AirflowError::InvalidUrl { .. }),
+            "got: {error:?}"
+        );
+    }
+
+    #[test]
+    fn accepts_a_valid_endpoint() {
+        let client = BaseClient::new(config("http://localhost:8080")).expect("should accept");
+        assert_eq!(client.endpoint().host_str(), Some("localhost"));
+    }
+
+    #[test]
+    fn preserves_a_reverse_proxy_prefix_when_building_requests() {
+        let client =
+            BaseClient::new(config("http://localhost:8080/airflow")).expect("should accept");
+        let url = client.endpoint().join("api/v2/dags").expect("should join");
+        assert_eq!(url.as_str(), "http://localhost:8080/airflow/api/v2/dags");
+    }
+}