This is an automated email from the ASF dual-hosted git repository.
amoghrajesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 13d7df70ecd Fix DbtCloudRunJobOperator having false failures during
deferred polling (#70581)
13d7df70ecd is described below
commit 13d7df70ecd11d6103e636ccd50f8cef0b335f45
Author: Amogh Desai <[email protected]>
AuthorDate: Thu Jul 30 10:25:16 2026 +0530
Fix DbtCloudRunJobOperator having false failures during deferred polling
(#70581)
---
.../src/airflow/providers/dbt/cloud/hooks/dbt.py | 45 +++++++--
.../cloud/tests/unit/dbt/cloud/hooks/test_dbt.py | 101 +++++++++++++++++++++
2 files changed, 138 insertions(+), 8 deletions(-)
diff --git a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/hooks/dbt.py
b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/hooks/dbt.py
index 89e512c51fb..b4ced28bbb7 100644
--- a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/hooks/dbt.py
+++ b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/hooks/dbt.py
@@ -41,7 +41,7 @@ from airflow.providers.http.hooks.http import HttpHook
if TYPE_CHECKING:
from requests.models import PreparedRequest, Response
- from airflow.models import Connection
+ from airflow.providers.common.compat.sdk import Connection
DBT_CAUSE_MAX_LENGTH = 255
@@ -254,12 +254,13 @@ class DbtCloudHook(HttpHook):
async def get_headers_tenants_from_connection(self) -> tuple[dict[str,
Any], str]:
"""Get Headers, tenants from the connection details."""
+ conn = await self._resolve_connection_async()
headers: dict[str, Any] = {}
- tenant = self._get_tenant_domain(self.connection)
+ tenant = self._get_tenant_domain(conn)
package_name, provider_version = _get_provider_info()
headers["User-Agent"] = f"{package_name}-v{provider_version}"
headers["Content-Type"] = "application/json"
- headers["Authorization"] = f"Token {self.connection.password}"
+ headers["Authorization"] = f"Token {conn.password}"
return headers, tenant
def _log_request_error(self, attempt_num: int, error: str) -> None:
@@ -307,7 +308,8 @@ class DbtCloudHook(HttpHook):
endpoint = f"{account_id}/runs/{run_id}/"
headers, tenant = await self.get_headers_tenants_from_connection()
url, params = self.get_request_url_params(tenant, endpoint,
include_related)
- proxies = self._get_proxies(self.connection) or {}
+ conn = await self._resolve_connection_async()
+ proxies = self._get_proxies(conn) or {}
proxy = proxies.get("https") if proxies and url.startswith("https")
else proxies.get("http")
extra_request_args = {}
@@ -341,13 +343,40 @@ class DbtCloudHook(HttpHook):
job_run_status: int = response["data"]["status"]
return job_run_status
+ @staticmethod
+ def _require_password(conn: Connection) -> Connection:
+ if not conn.password:
+ raise AirflowException("An API token is required to connect to dbt
Cloud.")
+ return conn
+
@cached_property
def connection(self) -> Connection:
- _connection = self.get_connection(self.dbt_cloud_conn_id)
- if not _connection.password:
- raise AirflowException("An API token is required to connect to dbt
Cloud.")
+ """
+ Resolve and cache the dbt Cloud connection (sync).
- return _connection # type: ignore[return-value]
+ Do not read this property from async code running inside the
triggerer's
+ event loop — it calls the synchronous ``get_connection()``, whose
secret-masking
+ step raises ``RuntimeError`` when invoked from a thread that's already
running an
+ event loop. Use ``_resolve_connection_async()`` instead; it shares
this property's
+ cache slot so the connection is still only looked up once per hook
instance
+ regardless of which path is used first.
+ """
+ return
self._require_password(self.get_connection(self.dbt_cloud_conn_id))
+
+ async def _resolve_connection_async(self) -> Connection:
+ """
+ Resolve and cache the dbt Cloud connection (async).
+
+ Shares the ``connection`` cached_property's cache slot so a connection
+ fetched from either the sync or async path is not looked up twice on
+ the same hook instance, and so the async path never touches the sync
+ ``get_connection()``/``mask_secret()`` path from inside a running
+ event loop (which raises in the triggerer).
+ """
+ if "connection" not in self.__dict__:
+ conn = await get_async_connection(self.dbt_cloud_conn_id)
+ self.__dict__["connection"] = self._require_password(conn)
+ return self.__dict__["connection"]
def get_conn(self, *args, **kwargs) -> Session:
tenant = self._get_tenant_domain(self.connection)
diff --git a/providers/dbt/cloud/tests/unit/dbt/cloud/hooks/test_dbt.py
b/providers/dbt/cloud/tests/unit/dbt/cloud/hooks/test_dbt.py
index 3ba9653bb0b..951e8505496 100644
--- a/providers/dbt/cloud/tests/unit/dbt/cloud/hooks/test_dbt.py
+++ b/providers/dbt/cloud/tests/unit/dbt/cloud/hooks/test_dbt.py
@@ -356,6 +356,107 @@ class TestDbtCloudHook:
assert mock_get_connection.call_count == 1
assert mock_get_async_connection.call_count == 0
+ @pytest.mark.asyncio
+ async def
test_get_headers_tenants_from_connection_does_not_use_sync_get_connection(self):
+ hook = DbtCloudHook(ACCOUNT_ID_CONN)
+
+ with (
+ patch.object(DbtCloudHook, "get_connection") as
mock_get_connection,
+ patch(
+ "airflow.providers.dbt.cloud.hooks.dbt.get_async_connection",
+ new=AsyncMock(
+ return_value=Connection(
+ conn_id=ACCOUNT_ID_CONN,
+ conn_type=DbtCloudHook.conn_type,
+ login=str(DEFAULT_ACCOUNT_ID),
+ password=TOKEN,
+ host=SINGLE_TENANT_DOMAIN,
+ )
+ ),
+ ) as mock_get_async_connection,
+ ):
+ headers, tenant = await hook.get_headers_tenants_from_connection()
+
+ assert tenant == SINGLE_TENANT_DOMAIN
+ assert headers["Authorization"] == f"Token {TOKEN}"
+ mock_get_connection.assert_not_called()
+ assert mock_get_async_connection.call_count == 1
+
+ @pytest.mark.asyncio
+ async def test_resolve_connection_cached_async(self):
+ hook = DbtCloudHook(ACCOUNT_ID_CONN)
+
+ with patch(
+ "airflow.providers.dbt.cloud.hooks.dbt.get_async_connection",
+ new=AsyncMock(
+ return_value=Connection(
+ conn_id=ACCOUNT_ID_CONN,
+ conn_type=DbtCloudHook.conn_type,
+ login=str(DEFAULT_ACCOUNT_ID),
+ password=TOKEN,
+ )
+ ),
+ ) as mock_get_async_connection:
+ first_call = await hook._resolve_connection_async()
+ second_call = await hook._resolve_connection_async()
+
+ assert first_call.password == TOKEN
+ assert second_call.password == TOKEN
+ assert mock_get_async_connection.call_count == 1
+
+ @pytest.mark.asyncio
+ async def test_connection_cache_shared_between_sync_and_async(self):
+ hook = DbtCloudHook(ACCOUNT_ID_CONN)
+
+ with (
+ patch.object(
+ DbtCloudHook,
+ "get_connection",
+ return_value=Connection(
+ conn_id=ACCOUNT_ID_CONN,
+ conn_type=DbtCloudHook.conn_type,
+ login=str(DEFAULT_ACCOUNT_ID),
+ password=TOKEN,
+ ),
+ ) as mock_get_connection,
+ patch(
+ "airflow.providers.dbt.cloud.hooks.dbt.get_async_connection",
+ new=AsyncMock(
+ return_value=Connection(
+ conn_id=ACCOUNT_ID_CONN,
+ conn_type=DbtCloudHook.conn_type,
+ login=str(DEFAULT_ACCOUNT_ID),
+ password=TOKEN,
+ )
+ ),
+ ) as mock_get_async_connection,
+ ):
+ sync_conn = hook.connection
+ async_conn = await hook._resolve_connection_async()
+
+ assert sync_conn.password == TOKEN
+ assert async_conn.password == TOKEN
+
+ assert mock_get_connection.call_count == 1
+ assert mock_get_async_connection.call_count == 0
+
+ @pytest.mark.asyncio
+ async def test_resolve_connection_async_requires_password(self):
+ hook = DbtCloudHook(ACCOUNT_ID_CONN)
+
+ with patch(
+ "airflow.providers.dbt.cloud.hooks.dbt.get_async_connection",
+ new=AsyncMock(
+ return_value=Connection(
+ conn_id=ACCOUNT_ID_CONN,
+ conn_type=DbtCloudHook.conn_type,
+ login=str(DEFAULT_ACCOUNT_ID),
+ )
+ ),
+ ):
+ with pytest.raises(AirflowException, match="An API token is
required"):
+ await hook._resolve_connection_async()
+
@pytest.mark.parametrize(
argnames=("conn_id", "account_id"),
argvalues=[(ACCOUNT_ID_CONN, None), (NO_ACCOUNT_ID_CONN, ACCOUNT_ID)],
