Migrate Airbyte provider to airbyte-api library 1.0 and httpx

providers/airbyte/README.rst

@@ -50,14 +50,14 @@ The package supports the following python versions: 3.10,3.11,3.12,3.13,3.14
 Requirements
 ------------
 
-==========================================  ===================
+==========================================  ==================
 PIP package                                 Version required
-==========================================  ===================
+==========================================  ==================
 ``apache-airflow``                          ``>=2.11.0``
 ``apache-airflow-providers-common-compat``  ``>=1.12.0``
-``airbyte-api``                             ``>=0.52.0,<1.0.0``
-``requests``                                ``>=2.32.0``
-==========================================  ===================
+``airbyte-api``                             ``>=1.0.0,<2.0``
+``httpx``                                   ``>=0.28.1``
+==========================================  ==================
 
 The changelog for the provider package can be found in the
 `changelog <https://airflow.apache.org/docs/apache-airflow-providers-airbyte/5.5.2/changelog.html>`_.

providers/airbyte/docs/changelog.rst

@@ -27,6 +27,26 @@ Changelog
 ---------
 
 
+Transitive dependency changes
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. note::
+  This release upgrades the provider's ``airbyte-api`` dependency to the ``1.x`` series and switches
+  the underlying HTTP client from ``requests`` to ``httpx``.
+
+  No changes are required for typical Dag authors: the ``AirbyteHook``, ``AirbyteTriggerSyncOperator``,
+  and ``AirbyteJobSensor`` public interfaces are unchanged, and the Airbyte connection configuration
+  (including the ``proxies`` extra) keeps the same format.
+
+  Action is required only if your environment relies on the provider's transitive dependencies:
+
+  * ``airbyte-api`` is now ``>=1.0.0,<2.0`` (previously ``>=0.52.0,<1.0.0``). The ``1.x`` SDK is built
+    on Pydantic models, so any code importing ``airbyte_api`` directly must pass request objects as
+    keyword arguments (e.g. ``GetJobRequest(job_id=...)``) and handle its stricter response validation.
+  * ``requests`` is no longer installed by this provider. If your code relied on it being pulled in
+    transitively, declare ``requests`` as an explicit dependency of your own project.
+
+
 5.5.2
 .....
 

providers/airbyte/docs/connections.rst

@@ -57,9 +57,8 @@ Client Secret (optional)
     Leave blank for Airbyte OSS deployments without auth enabled.
 
 Extra (optional)
-    Specify custom proxies in JSON format.
-    Following default requests parameters are taken into account:
+    Specify the ``proxies`` key in JSON format to route traffic through an HTTP proxy.
 
     * ``proxies``
 
-    Example: ``{"http": "http://proxy.example.com:8080", "https": "http://proxy.example.com:8080"}``
+    Example: ``{"proxies": {"http": "http://proxy.example.com:8080", "https": "http://proxy.example.com:8080"}}``

providers/airbyte/docs/index.rst

@@ -96,14 +96,14 @@ Requirements
 
 The minimum Apache Airflow version supported by this provider distribution is ``2.11.0``.
 
-==========================================  ===================
+==========================================  ==================
 PIP package                                 Version required
-==========================================  ===================
+==========================================  ==================
 ``apache-airflow``                          ``>=2.11.0``
 ``apache-airflow-providers-common-compat``  ``>=1.12.0``
-``airbyte-api``                             ``>=0.52.0,<1.0.0``
-``requests``                                ``>=2.32.0``
-==========================================  ===================
+``airbyte-api``                             ``>=1.0.0,<2.0``
+``httpx``                                   ``>=0.28.1``
+==========================================  ==================
 
 Downloading official packages
 -----------------------------

providers/airbyte/pyproject.toml

@@ -61,11 +61,8 @@ requires-python = ">=3.10"
 dependencies = [
     "apache-airflow>=2.11.0",
     "apache-airflow-providers-common-compat>=1.12.0",
-    # airbyte-api 1.0.0 migrated from requests to httpx; passing a requests.Session
-    # as the client now fails its HttpClient protocol check. Cap until the httpx
-    # migration lands; tracked at https://github.com/apache/airflow/pull/68882
-    "airbyte-api>=0.52.0,<1.0.0",
-    "requests>=2.32.0",
+    "airbyte-api>=1.0.0,<2.0",
+    "httpx>=0.28.1",
 ]
 
 [dependency-groups]

providers/airbyte/src/airflow/providers/airbyte/hooks/airbyte.py

@@ -20,10 +20,10 @@
 import time
 from typing import Any, TypeVar
 
+import httpx
 from airbyte_api import AirbyteAPI
 from airbyte_api.api import CancelJobRequest, GetJobRequest
 from airbyte_api.models import JobCreateRequest, JobStatusEnum, JobTypeEnum, SchemeClientCredentials, Security
-from requests import Session
 
 from airflow.providers.common.compat.sdk import AirflowException, BaseHook
 
@@ -110,8 +110,19 @@ def create_api_session(self) -> AirbyteAPI:
         client = None
         if self.conn["proxies"]:
             self.log.debug("Creating client proxy...")
-            client = Session()
-            client.proxies = self.conn["proxies"]
+            proxies = self.conn["proxies"]
+            mounts = {}
+            if isinstance(proxies, dict):
+                for scheme, proxy_url in proxies.items():
+                    # httpx mount keys require a "://" suffix
+                    key = scheme if "://" in scheme else f"{scheme}://"
+                    mounts[key] = httpx.HTTPTransport(proxy=proxy_url)
+            else:
+                mounts = {
+                    "http://": httpx.HTTPTransport(proxy=proxies),
+                    "https://": httpx.HTTPTransport(proxy=proxies),
+                }
+            client = httpx.Client(mounts=mounts)
 
         return AirbyteAPI(
             server_url=self.conn["host"],

providers/airbyte/tests/unit/airbyte/hooks/test_airbyte.py

@@ -19,6 +19,7 @@
 
 from unittest import mock
 
+import httpx
 import pytest
 from airbyte_api.api import CancelJobRequest, GetJobRequest
 from airbyte_api.models import JobResponse, JobStatusEnum, JobTypeEnum
@@ -86,7 +87,7 @@ def setup_connections(self, create_connection_without_db):
 
     def return_value_get_job(self, status):
         response = mock.Mock()
-        response.job_response = JobResponse(
+        response.job_response = JobResponse.model_construct(
             connection_id="connection-mock",
             job_id=self.job_id,
             start_time="today",
@@ -146,7 +147,7 @@ def test_cancel_job(self, cancel_job_mock):
     def test_wait_for_job_succeeded(self, mock_get_job):
         mock_get_job.side_effect = [self.return_value_get_job(JobStatusEnum.SUCCEEDED)]
         self.hook.wait_for_job(job_id=self.job_id, wait_seconds=0)
-        mock_get_job.assert_called_once_with(request=GetJobRequest(self.job_id))
+        mock_get_job.assert_called_once_with(request=GetJobRequest(job_id=self.job_id))
 
     @mock.patch("airbyte_api.jobs.Jobs.get_job")
     def test_wait_for_job_error(self, mock_get_job):
@@ -157,7 +158,10 @@ def test_wait_for_job_error(self, mock_get_job):
         with pytest.raises(AirflowException, match="Job failed"):
             self.hook.wait_for_job(job_id=self.job_id, wait_seconds=0)
 
-        calls = [mock.call(request=GetJobRequest(self.job_id)), mock.call(request=GetJobRequest(self.job_id))]
+        calls = [
+            mock.call(request=GetJobRequest(job_id=self.job_id)),
+            mock.call(request=GetJobRequest(job_id=self.job_id)),
+        ]
         mock_get_job.assert_has_calls(calls)
 
     @mock.patch("airbyte_api.jobs.Jobs.get_job")
@@ -168,7 +172,10 @@ def test_wait_for_job_incomplete_succeeded(self, mock_get_job):
         ]
         self.hook.wait_for_job(job_id=self.job_id, wait_seconds=0)
 
-        calls = [mock.call(request=GetJobRequest(self.job_id)), mock.call(request=GetJobRequest(self.job_id))]
+        calls = [
+            mock.call(request=GetJobRequest(job_id=self.job_id)),
+            mock.call(request=GetJobRequest(job_id=self.job_id)),
+        ]
         mock_get_job.assert_has_calls(calls)
 
     @mock.patch("airbyte_api.jobs.Jobs.get_job")
@@ -182,9 +189,9 @@ def test_wait_for_job_timeout(self, mock_cancel_job, mock_get_job):
             self.hook.wait_for_job(job_id=self.job_id, wait_seconds=2, timeout=1)
 
         get_calls = [
-            mock.call(request=GetJobRequest(self.job_id)),
+            mock.call(request=GetJobRequest(job_id=self.job_id)),
         ]
-        cancel_calls = [mock.call(request=CancelJobRequest(self.job_id))]
+        cancel_calls = [mock.call(request=CancelJobRequest(job_id=self.job_id))]
         mock_get_job.assert_has_calls(get_calls)
         mock_cancel_job.assert_has_calls(cancel_calls)
         assert mock_get_job.mock_calls == get_calls
@@ -199,7 +206,10 @@ def test_wait_for_job_state_unrecognized(self, mock_get_job):
         with pytest.raises(AirflowException, match="unexpected state"):
             self.hook.wait_for_job(job_id=self.job_id, wait_seconds=0)
 
-        calls = [mock.call(request=GetJobRequest(self.job_id)), mock.call(request=GetJobRequest(self.job_id))]
+        calls = [
+            mock.call(request=GetJobRequest(job_id=self.job_id)),
+            mock.call(request=GetJobRequest(job_id=self.job_id)),
+        ]
         mock_get_job.assert_has_calls(calls)
 
     @mock.patch("airbyte_api.jobs.Jobs.get_job")
@@ -211,7 +221,10 @@ def test_wait_for_job_cancelled(self, mock_get_job):
         with pytest.raises(AirflowException, match="Job was cancelled"):
             self.hook.wait_for_job(job_id=self.job_id, wait_seconds=0)
 
-        calls = [mock.call(request=GetJobRequest(self.job_id)), mock.call(request=GetJobRequest(self.job_id))]
+        calls = [
+            mock.call(request=GetJobRequest(job_id=self.job_id)),
+            mock.call(request=GetJobRequest(job_id=self.job_id)),
+        ]
         mock_get_job.assert_has_calls(calls)
 
     @mock.patch("airbyte_api.health.Health.get_health_check")
@@ -236,15 +249,18 @@ def test_connection_failure(self, mock_get_health_check):
         assert msg == '{"message": "internal server error"}'
 
     def test_create_api_session_with_proxy(self):
-        """
-        Test the creation of the API session with proxy settings.
-        """
-        # Create a new AirbyteHook instance
+        """Test that proxy settings produce an httpx.Client with per-scheme transport mounts."""
         hook = AirbyteHook(airbyte_conn_id=self.airbyte_conn_id_with_proxy)
 
-        # Check if the session is created correctly
         assert hook.airbyte_api is not None
-        assert hook.airbyte_api.sdk_configuration.client.proxies == self._mock_proxy["proxies"]
+        client = hook.airbyte_api.sdk_configuration.client
+        assert isinstance(client, httpx.Client)
+
+        default_transport = client._transport
+        for scheme in self._mock_proxy["proxies"]:
+            url = httpx.URL(f"{scheme}://example.com")
+            transport = client._transport_for_url(url)
+            assert transport is not default_transport, f"Expected proxy transport for {scheme}"
 
     def test_create_api_session_without_credentials(self):
         """Test that a session without OAuth credentials creates an unauthenticated client."""