feat: support Airflow 2 and 3 APIs

.env.example

@@ -1,3 +1,5 @@
 AIRFLOW_BASE_URL=http://localhost:8080
 AIRFLOW_USERNAME=your_username
 AIRFLOW_PASSWORD=your_password
+AIRFLOW_API_VERSION=v2
+AIRFLOW_AUTH_MODE=token

README.md

@@ -121,8 +121,15 @@ Configure:
 AIRFLOW_BASE_URL=http://localhost:8080
 AIRFLOW_USERNAME=your_username
 AIRFLOW_PASSWORD=your_password
+AIRFLOW_API_VERSION=v2
+AIRFLOW_AUTH_MODE=token
 ```
 
+Use `AIRFLOW_API_VERSION=v1` with `AIRFLOW_AUTH_MODE=basic` for Airflow 2.x
+Stable REST API deployments. Airflow 3.x uses the `v2` API and typically uses
+token authentication. Authentication still depends on the API auth backend
+configured in the Airflow deployment.
+
 Load the environment variables:
 
 ```bash
@@ -288,6 +295,7 @@ The current implementation should be considered experimental and is not yet inte
 Planned areas include:
 
 - DAG-level analysis models
+- richer CLI reporting
 - broader Airflow compatibility testing
 
 ## License

src/flowsense/config.py

@@ -9,6 +9,8 @@ class AirflowConfig:
     base_url: str
     username: str
     password: str
+    api_version: str = "v2"
+    auth_mode: str = "token"
 
 
 def get_airflow_config() -> AirflowConfig:
@@ -19,6 +21,8 @@ def get_airflow_config() -> AirflowConfig:
 
     username = os.getenv("AIRFLOW_USERNAME")
     password = os.getenv("AIRFLOW_PASSWORD")
+    api_version = os.getenv("AIRFLOW_API_VERSION", "v2")
+    auth_mode = os.getenv("AIRFLOW_AUTH_MODE", "token")
 
     if not username:
         raise RuntimeError("AIRFLOW_USERNAME environment variable is required.")
@@ -30,4 +34,6 @@ def get_airflow_config() -> AirflowConfig:
         base_url=base_url,
         username=username,
         password=password,
+        api_version=api_version,
+        auth_mode=auth_mode,
     )

src/flowsense/infrastructure/airflow/client.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Self
+from typing import Any, Literal, Self
 
 import httpx
 
@@ -18,6 +18,8 @@
 )
 
 PAGE_SIZE = 100
+AirflowApiVersion = Literal["v1", "v2"]
+AirflowAuthMode = Literal["basic", "token"]
 
 
 class AirflowClient:
@@ -26,13 +28,24 @@ def __init__(
         base_url: str | None = None,
         username: str | None = None,
         password: str | None = None,
+        api_version: AirflowApiVersion | None = None,
+        auth_mode: AirflowAuthMode | None = None,
         http_client: httpx.Client | None = None,
     ):
         config = get_airflow_config()
 
         self.base_url = (base_url or config.base_url).rstrip("/")
         self.username = username or config.username
         self.password = password or config.password
+        self.api_version = api_version or config.api_version
+        self.auth_mode = auth_mode or config.auth_mode
+
+        if self.api_version not in {"v1", "v2"}:
+            raise ValueError("api_version must be 'v1' or 'v2'")
+
+        if self.auth_mode not in {"basic", "token"}:
+            raise ValueError("auth_mode must be 'basic' or 'token'")
+
         self._owns_http_client = http_client is None
         self._http_client = http_client or httpx.Client(timeout=10.0)
         self._token: str | None = None
@@ -94,10 +107,21 @@ def _get_token(self) -> str:
         return token
 
     def _headers(self) -> dict[str, str]:
-        return {
-            "Authorization": f"Bearer {self._get_token()}",
-            "Accept": "application/json",
-        }
+        headers = {"Accept": "application/json"}
+
+        if self.auth_mode == "token":
+            headers["Authorization"] = f"Bearer {self._get_token()}"
+
+        return headers
+
+    def _authentication(self) -> httpx.BasicAuth | None:
+        if self.auth_mode == "basic":
+            return httpx.BasicAuth(self.username, self.password)
+
+        return None
+
+    def _api_url(self, path: str) -> str:
+        return f"{self.base_url}/api/{self.api_version}{path}"
 
     def _get_paginated(
         self,
@@ -109,11 +133,18 @@ def _get_paginated(
         last_page: dict = {}
 
         while True:
+            request_kwargs: dict[str, object] = {
+                "headers": self._headers(),
+                "params": {"limit": PAGE_SIZE, "offset": offset},
+            }
+            authentication = self._authentication()
+            if authentication is not None:
+                request_kwargs["auth"] = authentication
+
             response = self._request(
                 method="GET",
                 url=url,
-                headers=self._headers(),
-                params={"limit": PAGE_SIZE, "offset": offset},
+                **request_kwargs,
             )
 
             last_page = response.json()
@@ -140,7 +171,7 @@ def _get_paginated(
 
     def get_dag_runs(self, dag_id: str) -> dict:
         return self._get_paginated(
-            url=f"{self.base_url}/api/v2/dags/{dag_id}/dagRuns",
+            url=self._api_url(f"/dags/{dag_id}/dagRuns"),
             collection_key="dag_runs",
         )
 
@@ -150,10 +181,7 @@ def get_task_instances(
         dag_run_id: str,
     ) -> dict:
         return self._get_paginated(
-            url=(
-                f"{self.base_url}/api/v2/dags/{dag_id}/dagRuns/"
-                f"{dag_run_id}/taskInstances"
-            ),
+            url=self._api_url(f"/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances"),
             collection_key="task_instances",
         )
 
@@ -164,7 +192,7 @@ def collect_task_runs(self, dag_id: str) -> list[TaskRun]:
         ]
 
         def run_timestamp(run: AirflowDagRunDTO) -> float:
-            timestamp = run.run_after or run.queued_at
+            timestamp = run.run_after or run.logical_date or run.queued_at
             return timestamp.timestamp() if timestamp is not None else float("-inf")
 
         dag_runs.sort(key=run_timestamp)
@@ -198,7 +226,7 @@ def run_timestamp(run: AirflowDagRunDTO) -> float:
 
     def get_dag_tasks(self, dag_id: str) -> dict:
         return self._get_paginated(
-            url=f"{self.base_url}/api/v2/dags/{dag_id}/tasks",
+            url=self._api_url(f"/dags/{dag_id}/tasks"),
             collection_key="tasks",
         )
 

src/flowsense/infrastructure/airflow/dto.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import AliasChoices, BaseModel, ConfigDict, Field
 
 
 class AirflowDTO(BaseModel):
@@ -10,6 +10,10 @@ class AirflowDTO(BaseModel):
 class AirflowDagRunDTO(AirflowDTO):
     dag_run_id: str
     state: str | None = None
+    logical_date: datetime | None = Field(
+        default=None,
+        validation_alias=AliasChoices("logical_date", "execution_date"),
+    )
     run_after: datetime | None = None
     queued_at: datetime | None = None