Airflow 3.x API based connector (#26624)

* Add Airflow Connector with API integration

* Add Airflow Connector with API integration

* Update generated TypeScript types

* Add Airflow Connector with API integration improvements

* fix: username password flow for airflow 3, example yaml file, & sidebar docs

* fix type in UI

* Fix integration tests, fixed UI rendering and docs, improved OpenLineageResolver

* Fix pytests

* move connector

* Update generated TypeScript types

* fix: response parsing for astronomer airflow

* feat: added service account auth for airflow rest connection when composer managed airflow along with token

* fix: airflow rest api connection class converter and airflow.md

* feat: add mwaa config support for authentication

* s3 & column lineage

* Update generated TypeScript types

* fix: test airflow mwaa client

* fix: removed unused method, and extra code for parsing response

* fix: git pr checks

* fix: removed airflowapi integration tests that requires real host instance and added test with mocking

* fix test

* improve test coverage

* push coverage

* fix: gitar comments

* fix: removed redundant files

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: Keshav Mohta <68001229+keshavmohta09@users.noreply.github.com>
Co-authored-by: Keshav Mohta <keshavmohta09@gmail.com>
Co-authored-by: ulixius9 <mayursingal9@gmail.com>

Author: 4 people

docker/development/docker-compose.yml

@@ -512,6 +512,9 @@ services:
       AIRFLOW__CORE__EXECUTOR: LocalExecutor
       AIRFLOW__LOGGING__LOGGING_LEVEL: ${AIRFLOW_LOGGING_LEVEL:-DEBUG}
       AIRFLOW__OPENMETADATA_AIRFLOW_APIS__DAG_GENERATED_CONFIGS: "/opt/airflow/dag_generated_configs"
+      # OpenLineage transport config (optional - enable for lineage via OL)
+      # AIRFLOW__OPENLINEAGE__TRANSPORT: '{"type": "http", "url": "http://openmetadata-server:8585/api/v1/openlineage/", "endpoint": "lineage", "auth": {"type": "api_key", "api_key": "<OM_JWT_TOKEN>"}}'
+      # AIRFLOW__OPENLINEAGE__NAMESPACE: local_airflow
       DB_HOST: ${AIRFLOW_DB_HOST:-mysql}
       DB_PORT: ${AIRFLOW_DB_PORT:-3306}
       AIRFLOW_DB: ${AIRFLOW_DB:-airflow_db}

ingestion/src/metadata/clients/aws_client.py

@@ -43,6 +43,7 @@ class AWSServices(Enum):
     REDSHIFT = "redshift"
     REDSHIFT_SERVERLESS = "redshift-serverless"
     LAKE_FORMATION = "lakeformation"
+    MWAA = "mwaa"
 
 
 class AWSAssumeRoleException(Exception):
@@ -253,3 +254,6 @@ def get_redshift_client(self):
 
     def get_redshift_serverless_client(self):
         return self.get_client(AWSServices.REDSHIFT_SERVERLESS.value)
+
+    def get_mwaa_client(self):
+        return self.get_client(AWSServices.MWAA.value)

ingestion/src/metadata/ingestion/source/pipeline/airflow/api/__init__.py

@@ -0,0 +1,122 @@
+#  Copyright 2025 Collate
+#  Licensed under the Collate Community License, Version 1.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+"""
+Auth helper functions for the Airflow REST API client.
+"""
+import base64
+import traceback
+from datetime import datetime, timedelta, timezone
+from typing import Callable, Optional, Tuple
+
+import requests
+
+from metadata.utils.credentials import (
+    get_gcp_impersonate_credentials,
+    set_google_credentials,
+)
+from metadata.utils.logger import ingestion_logger
+
+logger = ingestion_logger()
+
+TokenCallback = Callable[[], Tuple[str, object]]
+
+_JWT_REFRESH_INTERVAL_SECONDS = (
+    25 * 60
+)  # re-fetch every 25 min, well within Airflow's ~30-60 min TTL
+_BASIC_AUTH_TTL_SECONDS = (
+    7 * 24 * 3600
+)  # basic auth doesn't expire; skip retry for 7 days
+
+
+def try_exchange_jwt(
+    host: str, username: str, password: str, verify: bool
+) -> Optional[str]:
+    """POST {host}/auth/token to get a JWT Bearer token (Airflow 3.x). Returns None on failure."""
+    try:
+        resp = requests.post(
+            f"{host}/auth/token",
+            json={"username": username, "password": password},
+            timeout=10,
+            verify=verify,
+        )
+        resp.raise_for_status()
+        return resp.json().get("access_token")
+    except Exception:
+        logger.debug(
+            "JWT token exchange failed (likely Airflow 2.x): %s", traceback.format_exc()
+        )
+        return None
+
+
+def build_access_token_callback(token: str) -> TokenCallback:
+    """Returns a static token callback with no expiry."""
+    return lambda: (token, 0)
+
+
+def build_basic_auth_callback(
+    host: str, username: str, password: str, verify: bool
+) -> Tuple[TokenCallback, None]:
+    """
+    Returns (callback, None). auth_token_mode=None means client.py uses the
+    token value as-is; the callback embeds 'Bearer' or 'Basic' prefix itself.
+
+    On every refresh cycle the callback re-calls try_exchange_jwt so the JWT
+    is always freshly issued — no stale-token 401s for long-running ingestions.
+    Falls back to Basic auth for Airflow 2.x servers.
+    """
+
+    def _callback() -> Tuple[str, object]:
+        jwt = try_exchange_jwt(host, username, password, verify)
+        if jwt:
+            return f"Bearer {jwt}", _JWT_REFRESH_INTERVAL_SECONDS
+        b64 = base64.b64encode(f"{username}:{password}".encode()).decode()
+        return f"Basic {b64}", _BASIC_AUTH_TTL_SECONDS
+
+    return _callback, None
+
+
+def build_gcp_token_callback(gcp_credentials) -> TokenCallback:
+    """
+    Returns a token callback that fetches and auto-refreshes GCP OAuth2 tokens.
+
+    Supports all 4 GCP credential types via set_google_credentials():
+    - GcpCredentialsValues: service account JSON values (clientEmail, privateKey, etc.)
+    - GcpCredentialsPath: path to a credentials JSON file
+    - GcpExternalAccount: workload identity federation
+    - GcpADC: application default credentials
+
+    Also handles optional service account impersonation via gcpImpersonateServiceAccount.
+    """
+    set_google_credentials(gcp_credentials)
+    impersonate = gcp_credentials.gcpImpersonateServiceAccount
+
+    def _callback() -> Tuple[str, datetime]:
+        import google.auth
+        from google.auth.transport.requests import Request as AuthRequest
+
+        if impersonate and impersonate.impersonateServiceAccount:
+            credentials = get_gcp_impersonate_credentials(
+                impersonate_service_account=impersonate.impersonateServiceAccount,
+                scopes=["https://www.googleapis.com/auth/cloud-platform"],
+                lifetime=impersonate.lifetime,
+            )
+        else:
+            credentials, _ = google.auth.default(
+                scopes=["https://www.googleapis.com/auth/cloud-platform"]
+            )
+
+        credentials.refresh(AuthRequest())
+        expiry = getattr(credentials, "expiry", None) or (
+            datetime.now(timezone.utc) + timedelta(minutes=55)
+        )
+        return (credentials.token, expiry)
+
+    return _callback