You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2026/05/10 02:14:16 UTC
(airflow) branch main updated: Add to list dbt Cloud job runs with support for filtering by status, ordering, and job_definition_id, and optionally returning only the most recent run via latest_only. Unit tests have been included. (#66150)
This is an automated email from the ASF dual-hosted git repository.
potiuk 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 50a115299d5 Add to list dbt Cloud job runs with support for filtering by status, ordering, and job_definition_id, and optionally returning only the most recent run via latest_only. Unit tests have been included. (#66150)
50a115299d5 is described below
commit 50a115299d543a19f52c3d6a926962949642249c
Author: SameerMesiah97 <75...@users.noreply.github.com>
AuthorDate: Sun May 10 03:14:07 2026 +0100
Add to list dbt Cloud job runs with support for filtering by status, ordering, and job_definition_id, and optionally returning only the most recent run via latest_only. Unit tests have been included. (#66150)
Co-authored-by: Sameer Mesiah <sm...@gmail.com>
---
providers/dbt/cloud/docs/operators.rst | 21 +++
.../airflow/providers/dbt/cloud/operators/dbt.py | 91 ++++++++++
.../tests/system/dbt/cloud/example_dbt_cloud.py | 9 +-
.../tests/unit/dbt/cloud/operators/test_dbt.py | 201 +++++++++++++++++++++
4 files changed, 321 insertions(+), 1 deletion(-)
diff --git a/providers/dbt/cloud/docs/operators.rst b/providers/dbt/cloud/docs/operators.rst
index fe4d7b06619..a765035ef9f 100644
--- a/providers/dbt/cloud/docs/operators.rst
+++ b/providers/dbt/cloud/docs/operators.rst
@@ -162,3 +162,24 @@ For more information on dbt Cloud list jobs, reference
:dedent: 4
:start-after: [START howto_operator_dbt_cloud_list_jobs]
:end-before: [END howto_operator_dbt_cloud_list_jobs]
+
+
+.. _howto/operator:DbtCloudListJobRunsOperator:
+
+List job runs
+~~~~~~~~~~~~~
+
+Use the :class:`~airflow.providers.dbt.cloud.operators.dbt.DbtCloudListJobRunsOperator` to list
+all job runs tied to a specified dbt Cloud account. The ``account_id`` must be supplied either
+through the connection or supplied as a parameter to the task.
+
+If a ``job_id`` is supplied, only job runs pertaining to this job will be retrieved.
+
+For more information on dbt Cloud list jobs, reference
+`this documentation <https://docs.getdbt.com/dbt-cloud/api-v2#tag/Jobs/operation/listJobRunsForAccount>`__.
+
+.. exampleinclude:: /../tests/system/dbt/cloud/example_dbt_cloud.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_dbt_cloud_list_job_runs]
+ :end-before: [END howto_operator_dbt_cloud_list_job_runs]
diff --git a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py
index 2fff2728abe..0718b1f5abd 100644
--- a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py
+++ b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py
@@ -463,3 +463,94 @@ class DbtCloudListJobsOperator(BaseOperator):
buffer.append(job["id"])
self.log.info("Jobs in the specified dbt Cloud account are: %s", ", ".join(map(str, buffer)))
return buffer
+
+
+class DbtCloudListJobRunsOperator(BaseOperator):
+ """
+ List job runs in dbt Cloud.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:DbtCloudListJobRunsOperator`
+
+ Retrieves metadata for job runs tied to a specified dbt Cloud account.
+ Optionally filters by job_id and allows ordering.
+
+ :param dbt_cloud_conn_id: The connection ID for connecting to dbt Cloud.
+ :param account_id: Optional. If not provided, the account ID from the connection is used.
+ :param job_id: Optional. Filter runs for a specific job.
+ :param order_by: Optional. Field to order results by (e.g. "-id").
+ :param include_related: Optional. Related fields to include (e.g. ["job", "environment"]).
+ :param hook_params: Extra arguments passed to the DbtCloudHook constructor.
+ :param latest_only: If True, return only the most recent job run.
+ :param status_filter: Optional. Filter runs by status code(s).
+ Accepts an integer or a sequence of integers.
+ """
+
+ template_fields = (
+ "account_id",
+ "job_id",
+ "order_by",
+ "include_related",
+ )
+
+ def __init__(
+ self,
+ *,
+ dbt_cloud_conn_id: str = DbtCloudHook.default_conn_name,
+ account_id: int | None = None,
+ job_id: int | None = None,
+ order_by: str | None = None,
+ include_related: list[str] | None = None,
+ hook_params: dict[str, Any] | None = None,
+ latest_only: bool = False,
+ status_filter: int | list[int] | None = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.dbt_cloud_conn_id = dbt_cloud_conn_id
+ self.account_id = account_id
+ self.job_id = job_id
+ self.order_by = order_by
+ self.include_related = include_related
+ self.hook_params = hook_params or {}
+ self.latest_only = latest_only
+ self.status_filter = status_filter
+
+ def execute(self, context) -> list[dict[str, Any]] | dict[str, Any] | None:
+ hook = DbtCloudHook(self.dbt_cloud_conn_id, **self.hook_params)
+
+ order_by = self.order_by or ("-created_at" if self.latest_only else None)
+
+ responses = hook.list_job_runs(
+ account_id=self.account_id,
+ include_related=self.include_related,
+ job_definition_id=self.job_id,
+ order_by=order_by,
+ )
+
+ runs: list[dict[str, Any]] = []
+ for response in responses:
+ runs.extend(response.json()["data"])
+
+ if self.status_filter is not None:
+ allowed = {self.status_filter} if isinstance(self.status_filter, int) else set(self.status_filter)
+
+ runs = [run for run in runs if run.get("status") is not None and int(run["status"]) in allowed]
+
+ if self.latest_only:
+ latest = runs[0] if runs else None
+
+ if latest:
+ self.log.info(
+ "Returning latest job run (id=%s, status=%s)",
+ latest["id"],
+ latest.get("status"),
+ )
+ else:
+ self.log.info("No job runs found for the given filters.")
+
+ return latest
+
+ self.log.info("Retrieved %s job runs.", len(runs))
+ return runs
diff --git a/providers/dbt/cloud/tests/system/dbt/cloud/example_dbt_cloud.py b/providers/dbt/cloud/tests/system/dbt/cloud/example_dbt_cloud.py
index 8a0ebad5611..a70a3042750 100644
--- a/providers/dbt/cloud/tests/system/dbt/cloud/example_dbt_cloud.py
+++ b/providers/dbt/cloud/tests/system/dbt/cloud/example_dbt_cloud.py
@@ -21,6 +21,7 @@ from datetime import datetime
from airflow.models import DAG
from airflow.providers.dbt.cloud.operators.dbt import (
DbtCloudGetJobRunArtifactOperator,
+ DbtCloudListJobRunsOperator,
DbtCloudListJobsOperator,
DbtCloudRunJobOperator,
)
@@ -94,9 +95,15 @@ with DAG(
list_dbt_jobs = DbtCloudListJobsOperator(task_id="list_dbt_jobs", account_id=106277, project_id=160645)
# [END howto_operator_dbt_cloud_list_jobs]
+ # [START howto_operator_dbt_cloud_list_job_runs]
+ list_dbt_job_runs = DbtCloudListJobRunsOperator(
+ task_id="list_dbt_job_runs", account_id=106277, job_id=160645
+ )
+ # [END howto_operator_dbt_cloud_list_job_runs]
+
begin >> Label("No async wait") >> trigger_job_run1
begin >> Label("Do async wait with sensor") >> trigger_job_run2
- [get_run_results_artifact, job_run_sensor, list_dbt_jobs] >> end
+ [get_run_results_artifact, job_run_sensor, list_dbt_jobs, list_dbt_job_runs] >> end
# Task dependency created via `XComArgs`:
# trigger_job_run1 >> get_run_results_artifact
diff --git a/providers/dbt/cloud/tests/unit/dbt/cloud/operators/test_dbt.py b/providers/dbt/cloud/tests/unit/dbt/cloud/operators/test_dbt.py
index 784cb62695e..e5df84277a1 100644
--- a/providers/dbt/cloud/tests/unit/dbt/cloud/operators/test_dbt.py
+++ b/providers/dbt/cloud/tests/unit/dbt/cloud/operators/test_dbt.py
@@ -27,6 +27,7 @@ from airflow.providers.common.compat.sdk import AirflowException, TaskDeferred,
from airflow.providers.dbt.cloud.hooks.dbt import DbtCloudHook, DbtCloudJobRunException, DbtCloudJobRunStatus
from airflow.providers.dbt.cloud.operators.dbt import (
DbtCloudGetJobRunArtifactOperator,
+ DbtCloudListJobRunsOperator,
DbtCloudListJobsOperator,
DbtCloudRunJobOperator,
)
@@ -933,3 +934,203 @@ class TestDbtCloudListJobsOperator:
mock_list_jobs.return_value.json.return_value = {}
operator.execute(context=self.mock_context)
mock_list_jobs.assert_called_once_with(account_id=account_id, order_by=None, project_id=PROJECT_ID)
+
+
+class TestDbtCloudListJobRunsOperator:
+ @patch("airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.list_job_runs")
+ @pytest.mark.parametrize(
+ ("conn_id", "account_id", "job_id"),
+ [
+ (ACCOUNT_ID_CONN, None, JOB_ID),
+ (NO_ACCOUNT_ID_CONN, ACCOUNT_ID, JOB_ID),
+ (DbtCloudHook.default_conn_name, None, None),
+ ],
+ )
+ def test_execute_list_job_runs(self, mock_list_job_runs, conn_id, account_id, job_id):
+ operator = DbtCloudListJobRunsOperator(
+ task_id=TASK_ID,
+ dbt_cloud_conn_id=conn_id,
+ account_id=account_id,
+ job_id=job_id,
+ )
+
+ mock_list_job_runs.return_value = [mock_response_json({"data": [{"id": 1}, {"id": 2}]})]
+
+ result = operator.execute(context={})
+
+ mock_list_job_runs.assert_called_once_with(
+ account_id=account_id,
+ include_related=None,
+ job_definition_id=job_id,
+ order_by=None,
+ )
+
+ assert result == [{"id": 1}, {"id": 2}]
+
+ @patch("airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.list_job_runs")
+ def test_execute_without_job_id(self, mock_list_job_runs):
+ operator = DbtCloudListJobRunsOperator(task_id=TASK_ID)
+
+ mock_list_job_runs.return_value = [mock_response_json({"data": [{"id": 1}, {"id": 2}]})]
+
+ result = operator.execute(context={})
+
+ mock_list_job_runs.assert_called_once_with(
+ account_id=None,
+ include_related=None,
+ job_definition_id=None,
+ order_by=None,
+ )
+
+ assert result == [{"id": 1}, {"id": 2}]
+
+ @patch("airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.list_job_runs")
+ @pytest.mark.parametrize(
+ ("latest_only", "status_filter", "response_data", "expected"),
+ [
+ # latest_only=True, empty.
+ (
+ True,
+ None,
+ [],
+ None,
+ ),
+ # status_filter and latest_only = True, empty.
+ (
+ True,
+ 10,
+ [{"id": 1, "status": 20}],
+ None,
+ ),
+ # status_filter and latest_only = False, empty.
+ (
+ False,
+ 10,
+ [{"id": 1, "status": 20}],
+ [],
+ ),
+ # status_filter single.
+ (
+ False,
+ 10,
+ [{"id": 1, "status": 10}, {"id": 2, "status": 20}],
+ [{"id": 1, "status": 10}],
+ ),
+ # status_filter multiple.
+ (
+ False,
+ [10, 20],
+ [
+ {"id": 1, "status": 10},
+ {"id": 2, "status": 20},
+ {"id": 3, "status": 30},
+ ],
+ [
+ {"id": 1, "status": 10},
+ {"id": 2, "status": 20},
+ ],
+ ),
+ # latest_only and status_filter.
+ (
+ True,
+ 10,
+ [
+ {"id": 1, "status": 20},
+ {"id": 2, "status": 10},
+ {"id": 3, "status": 10},
+ ],
+ {"id": 2, "status": 10},
+ ),
+ # "status" is string.
+ (
+ False,
+ 10,
+ [{"id": 1, "status": "10"}, {"id": 2, "status": "20"}],
+ [{"id": 1, "status": "10"}],
+ ),
+ # Missing status.
+ (
+ False,
+ 10,
+ [{"id": 1}, {"id": 2, "status": 10}],
+ [{"id": 2, "status": 10}],
+ ),
+ # 'None' status.
+ (
+ False,
+ 10,
+ [{"id": 1, "status": None}, {"id": 2, "status": 10}],
+ [{"id": 2, "status": 10}],
+ ),
+ ],
+ )
+ def test_execute_filtering_and_latest(
+ self,
+ mock_list_job_runs,
+ latest_only,
+ status_filter,
+ response_data,
+ expected,
+ ):
+ operator = DbtCloudListJobRunsOperator(
+ task_id=TASK_ID,
+ latest_only=latest_only,
+ status_filter=status_filter,
+ )
+
+ mock_list_job_runs.return_value = (
+ [mock_response_json({"data": response_data})] if response_data else []
+ )
+
+ result = operator.execute(context={})
+
+ assert result == expected
+
+ @patch("airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.list_job_runs")
+ @pytest.mark.parametrize(
+ ("latest_only", "order_by", "expected_order"),
+ [
+ (True, None, "-created_at"),
+ (True, "-id", "-id"), # user override should win
+ (False, None, None),
+ ],
+ )
+ def test_execute_order_by_behavior(
+ self,
+ mock_list_job_runs,
+ latest_only,
+ order_by,
+ expected_order,
+ ):
+ operator = DbtCloudListJobRunsOperator(
+ task_id=TASK_ID,
+ latest_only=latest_only,
+ order_by=order_by,
+ )
+
+ mock_list_job_runs.return_value = []
+
+ operator.execute(context={})
+
+ mock_list_job_runs.assert_called_once_with(
+ account_id=None,
+ include_related=None,
+ job_definition_id=None,
+ order_by=expected_order,
+ )
+
+ @patch("airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.list_job_runs")
+ def test_execute_multiple_pages(self, mock_list_job_runs):
+ operator = DbtCloudListJobRunsOperator(task_id=TASK_ID)
+
+ mock_list_job_runs.return_value = [
+ mock_response_json({"data": [{"id": 1, "status": 10}]}),
+ mock_response_json({"data": [{"id": 2, "status": 20}]}),
+ ]
+
+ result = operator.execute(context={})
+
+ assert result == [
+ {"id": 1, "status": 10},
+ {"id": 2, "status": 20},
+ ]