Validate dbt cause field to be less than 255 characters (apache#38896)

* Validate dbt `trigger_reason` field to be less than 255 characters

* Rename function to better reflect the behavior

* Move validation of cause parameter to DbtCloudHook

* Add stacklevel to dbt `cause` warning

airflow/providers/dbt/cloud/hooks/dbt.py

@@ -19,6 +19,7 @@
 import asyncio
 import json
 import time
+import warnings
 from enum import Enum
 from functools import cached_property, wraps
 from inspect import signature
@@ -39,6 +40,8 @@
 
     from airflow.models import Connection
 
+DBT_CAUSE_MAX_LENGTH = 255
+
 
 def fallback_to_default_account(func: Callable) -> Callable:
     """
@@ -420,6 +423,15 @@ def trigger_job_run(
         if additional_run_config is None:
             additional_run_config = {}
 
+        if cause is not None and len(cause) > DBT_CAUSE_MAX_LENGTH:
+            warnings.warn(
+                f"Cause `{cause}` exceeds limit of {DBT_CAUSE_MAX_LENGTH}"
+                f" characters and will be truncated.",
+                UserWarning,
+                stacklevel=2,
+            )
+            cause = cause[:DBT_CAUSE_MAX_LENGTH]
+
         payload = {
             "cause": cause,
             "steps_override": steps_override,

tests/providers/dbt/cloud/hooks/test_dbt.py

@@ -25,6 +25,7 @@
 from airflow.exceptions import AirflowException
 from airflow.models.connection import Connection
 from airflow.providers.dbt.cloud.hooks.dbt import (
+    DBT_CAUSE_MAX_LENGTH,
     DbtCloudHook,
     DbtCloudJobRunException,
     DbtCloudJobRunStatus,
@@ -394,6 +395,35 @@ def test_trigger_job_run_with_additional_run_configs(
         )
         hook._paginate.assert_not_called()
 
+    @pytest.mark.parametrize(
+        argnames="conn_id, account_id",
+        argvalues=[(ACCOUNT_ID_CONN, None), (NO_ACCOUNT_ID_CONN, ACCOUNT_ID)],
+        ids=["default_account", "explicit_account"],
+    )
+    @patch.object(DbtCloudHook, "run")
+    @patch.object(DbtCloudHook, "_paginate")
+    def test_trigger_job_run_with_longer_cause(self, mock_http_run, mock_paginate, conn_id, account_id):
+        hook = DbtCloudHook(conn_id)
+        cause = "Some cause that is longer than limit. " * 15
+        expected_cause = cause[:DBT_CAUSE_MAX_LENGTH]
+        assert len(cause) > DBT_CAUSE_MAX_LENGTH
+
+        with pytest.warns(
+            UserWarning,
+            match=f"Cause `{cause}` exceeds limit of {DBT_CAUSE_MAX_LENGTH}"
+            f" characters and will be truncated.",
+        ):
+            hook.trigger_job_run(job_id=JOB_ID, cause=cause, account_id=account_id)
+
+        assert hook.method == "POST"
+
+        _account_id = account_id or DEFAULT_ACCOUNT_ID
+        hook.run.assert_called_once_with(
+            endpoint=f"api/v2/accounts/{_account_id}/jobs/{JOB_ID}/run/",
+            data=json.dumps({"cause": expected_cause, "steps_override": None, "schema_override": None}),
+        )
+        hook._paginate.assert_not_called()
+
     @pytest.mark.parametrize(
         argnames="conn_id, account_id",
         argvalues=[(ACCOUNT_ID_CONN, None), (NO_ACCOUNT_ID_CONN, ACCOUNT_ID)],