Derive DbtCloudGetJobRunArtifactOperator default file name after rendering

providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py

@@ -410,10 +410,14 @@ def __init__(
         self.path = path
         self.account_id = account_id
         self.step = step
-        self.output_file_name = output_file_name or f"{self.run_id}_{self.path}".replace("/", "-")
+        self.output_file_name = output_file_name
         self.hook_params = hook_params or {}
 
     def execute(self, context: Context) -> str:
+        # run_id/path are template fields; derive the default name after rendering so it uses the
+        # resolved values, not the Jinja expressions.
+        if not self.output_file_name:
+            self.output_file_name = f"{self.run_id}_{self.path}".replace("/", "-")
         hook = DbtCloudHook(self.dbt_cloud_conn_id, **self.hook_params)
         response = hook.get_job_run_artifact(
             run_id=self.run_id, path=self.path, account_id=self.account_id, step=self.step

providers/dbt/cloud/tests/unit/dbt/cloud/operators/test_dbt.py

@@ -1058,6 +1058,27 @@ def test_get_text_artifact_with_step(self, mock_get_artifact, conn_id, account_i
         assert os.path.exists(tmp_path / operator.output_file_name)
         assert return_value == operator.output_file_name
 
+    @patch("airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.get_job_run_artifact")
+    def test_default_output_file_name_uses_rendered_path(self, mock_get_artifact, tmp_path, monkeypatch):
+        # The default name flattens "/"->"-", so a templated path must resolve before that.
+        operator = DbtCloudGetJobRunArtifactOperator(
+            task_id=TASK_ID,
+            dbt_cloud_conn_id=ACCOUNT_ID_CONN,
+            run_id=RUN_ID,
+            path="{{ params.p }}",
+            params={"p": "path/to/my/manifest.json"},
+            dag=self.dag,
+        )
+        operator.render_template_fields({"params": {"p": "path/to/my/manifest.json"}})
+
+        mock_get_artifact.return_value.json.return_value = {"data": "file contents"}
+        with monkeypatch.context() as ctx:
+            ctx.chdir(tmp_path)
+            return_value = operator.execute(context={})
+
+        assert operator.output_file_name == f"{RUN_ID}_path-to-my-manifest.json"
+        assert return_value == operator.output_file_name
+
     @patch("airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.get_job_run_artifact")
     @pytest.mark.parametrize(
         ("conn_id", "account_id"),

scripts/ci/prek/validate_operators_init_exemptions.txt

@@ -41,7 +41,6 @@ providers/databricks/src/airflow/providers/databricks/operators/databricks_repos
 providers/databricks/src/airflow/providers/databricks/operators/databricks_repos.py::DatabricksReposUpdateOperator
 providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py::DatabricksCopyIntoOperator
 providers/databricks/src/airflow/providers/databricks/sensors/databricks.py::DatabricksSQLStatementsSensor
-providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py::DbtCloudGetJobRunArtifactOperator
 providers/docker/src/airflow/providers/docker/operators/docker.py::DockerOperator
 providers/google/src/airflow/providers/google/cloud/operators/bigquery.py::BigQueryInsertJobOperator
 providers/google/src/airflow/providers/google/cloud/operators/cloud_batch.py::CloudBatchSubmitJobOperator