feat(serialization): implement truncation logic for rendered values

airflow-core/pyproject.toml

@@ -243,6 +243,7 @@ exclude = [
 "../shared/listeners/src/airflow_shared/listeners" = "src/airflow/_shared/listeners"
 "../shared/plugins_manager/src/airflow_shared/plugins_manager" = "src/airflow/_shared/plugins_manager"
 "../shared/providers_discovery/src/airflow_shared/providers_discovery" = "src/airflow/_shared/providers_discovery"
+"../shared/template_rendering/src/airflow_shared/template_rendering" = "src/airflow/_shared/template_rendering"
 
 [tool.hatch.build.targets.custom]
 path = "./hatch_build.py"

airflow-core/src/airflow/_shared/template_rendering

@@ -0,0 +1 @@
+../../../../shared/template_rendering/src/airflow_shared/template_rendering

airflow-core/src/airflow/serialization/helpers.py

@@ -23,6 +23,7 @@
 
 from airflow._shared.module_loading import qualname
 from airflow._shared.secrets_masker import redact
+from airflow._shared.template_rendering import truncate_rendered_value
 from airflow.configuration import conf
 from airflow.settings import json
 
@@ -83,10 +84,7 @@ def sort_dict_recursively(obj: Any) -> Any:
                 serialized = str(template_field)
         if len(serialized) > max_length:
             rendered = redact(serialized, name)
-            return (
-                "Truncated. You can change this behaviour in [core]max_templated_field_length. "
-                f"{rendered[: max_length - 79]!r}... "
-            )
+            return truncate_rendered_value(str(rendered), max_length)
         return serialized
     if not template_field and not isinstance(template_field, tuple):
         # Avoid unnecessary serialization steps for empty fields unless they are tuples
@@ -100,10 +98,7 @@ def sort_dict_recursively(obj: Any) -> Any:
     serialized = str(template_field)
     if len(serialized) > max_length:
         rendered = redact(serialized, name)
-        return (
-            "Truncated. You can change this behaviour in [core]max_templated_field_length. "
-            f"{rendered[: max_length - 79]!r}... "
-        )
+        return truncate_rendered_value(str(rendered), max_length)
     return template_field
 
 

airflow-core/tests/unit/models/test_renderedtifields.py

@@ -30,6 +30,7 @@
 from sqlalchemy import select
 
 from airflow import settings
+from airflow._shared.template_rendering import truncate_rendered_value
 from airflow._shared.timezones.timezone import datetime
 from airflow.configuration import conf
 from airflow.models import DagRun
@@ -124,13 +125,14 @@ def teardown_method(self):
             pytest.param(datetime(2018, 12, 6, 10, 55), "2018-12-06 10:55:00+00:00", id="datetime"),
             pytest.param(
                 "a" * 5000,
-                f"Truncated. You can change this behaviour in [core]max_templated_field_length. {('a' * 5000)[: max_length - 79]!r}... ",
+                truncate_rendered_value("a" * 5000, conf.getint("core", "max_templated_field_length")),
                 id="large_string",
             ),
             pytest.param(
                 LargeStrObject(),
-                f"Truncated. You can change this behaviour in "
-                f"[core]max_templated_field_length. {str(LargeStrObject())[: max_length - 79]!r}... ",
+                truncate_rendered_value(
+                    str(LargeStrObject()), conf.getint("core", "max_templated_field_length")
+                ),
                 id="large_object",
             ),
         ],

airflow-core/tests/unit/serialization/test_helpers.py

@@ -0,0 +1,31 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+
+def test_serialize_template_field_with_very_small_max_length(monkeypatch):
+    """Test that truncation message is prioritized even for very small max_length."""
+    monkeypatch.setenv("AIRFLOW__CORE__MAX_TEMPLATED_FIELD_LENGTH", "1")
+
+    from airflow.serialization.helpers import serialize_template_field
+
+    result = serialize_template_field("This is a long string", "test")
+
+    # The truncation message should be shown even if it exceeds max_length
+    # This ensures users always see why content is truncated
+    assert result
+    assert "Truncated. You can change this behaviour" in result

pyproject.toml

@@ -1314,6 +1314,7 @@ dev = [
     "apache-airflow-shared-providers-discovery",
     "apache-airflow-shared-secrets-backend",
     "apache-airflow-shared-secrets-masker",
+    "apache-airflow-shared-template-rendering",
     "apache-airflow-shared-timezones",
 ]
 
@@ -1373,6 +1374,7 @@ apache-airflow-shared-plugins-manager = { workspace = true }
 apache-airflow-shared-providers-discovery = { workspace = true }
 apache-airflow-shared-secrets-backend = { workspace = true }
 apache-airflow-shared-secrets-masker = { workspace = true }
+apache-airflow-shared-template-rendering = { workspace = true }
 apache-airflow-shared-timezones = { workspace = true }
 # Automatically generated provider workspace items (update_airflow_pyproject_toml.py)
 apache-airflow-providers-airbyte = { workspace = true }
@@ -1503,6 +1505,7 @@ members = [
     "shared/providers_discovery",
     "shared/secrets_backend",
     "shared/secrets_masker",
+    "shared/template_rendering",
     "shared/timezones",
     # Automatically generated provider workspace members (update_airflow_pyproject_toml.py)
     "providers/airbyte",

shared/template_rendering/pyproject.toml

@@ -0,0 +1,46 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[project]
+name = "apache-airflow-shared-template-rendering"
+description = "Shared template rendering utilities for Airflow distributions"
+version = "0.0"
+classifiers = [
+    "Private :: Do Not Upload",
+]
+
+dependencies = []
+
+[dependency-groups]
+dev = [
+    "apache-airflow-devel-common",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/airflow_shared"]
+
+[tool.ruff]
+extend = "../../pyproject.toml"
+src = ["src"]
+
+[tool.ruff.lint.per-file-ignores]
+# Ignore Doc rules et al for anything outside of tests
+"!src/*" = ["D", "S101", "TRY002"]

shared/template_rendering/src/airflow_shared/template_rendering/__init__.py

@@ -0,0 +1,86 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import logging
+
+log = logging.getLogger(__name__)
+
+# Public truncation configuration used by ``truncate_rendered_value``.
+# Exposed as module-level constants to avoid duplicating literals in callers/tests.
+TRUNCATE_MIN_CONTENT_LENGTH = 7
+TRUNCATE_PREFIX = "Truncated. You can change this behaviour in [core]max_templated_field_length. "
+TRUNCATE_SUFFIX = "..."
+
+
+def truncate_rendered_value(rendered: str, max_length: int) -> str:
+    """
+    Truncate a rendered template value to approximately ``max_length`` characters.
+
+    Behavior:
+
+    * If ``max_length <= 0``, an empty string is returned.
+    * A fixed prefix (``TRUNCATE_PREFIX``) and suffix (``TRUNCATE_SUFFIX``) are always
+      included when truncation occurs. The minimal truncation-only message is::
+
+          f"{TRUNCATE_PREFIX}{TRUNCATE_SUFFIX}"
+
+    * If ``max_length`` is smaller than the length of this truncation-only message, that
+      message is returned in full, even though its length may exceed ``max_length``.
+    * Otherwise, space remaining after the prefix and suffix is allocated to the original
+      ``rendered`` content. Content is only appended if at least
+      ``TRUNCATE_MIN_CONTENT_LENGTH`` characters are available; if fewer are available,
+      the truncation-only message is returned instead.
+
+    Note: this function is best-effort — the return value is intended to be no longer than
+    ``max_length``, but when ``max_length < len(TRUNCATE_PREFIX + TRUNCATE_SUFFIX)`` it
+    intentionally returns a longer string to preserve the full truncation message.
+    """
+    if max_length <= 0:
+        return ""
+
+    trunc_only = f"{TRUNCATE_PREFIX}{TRUNCATE_SUFFIX}"
+
+    if max_length < len(trunc_only):
+        return trunc_only
+
+    # Compute available space for content
+    overhead = len(TRUNCATE_PREFIX) + len(TRUNCATE_SUFFIX)
+    available = max_length - overhead
+
+    if available < TRUNCATE_MIN_CONTENT_LENGTH:
+        return trunc_only
+
+    # Slice content to fit and construct final string
+    content = rendered[:available].rstrip()
+    result = f"{TRUNCATE_PREFIX}{content}{TRUNCATE_SUFFIX}"
+
+    if len(result) > max_length:
+        log.warning(
+            "Truncated value still exceeds max_length=%d; this should not happen.",
+            max_length,
+        )
+
+    return result
+
+
+__all__ = [
+    "TRUNCATE_MIN_CONTENT_LENGTH",
+    "TRUNCATE_PREFIX",
+    "TRUNCATE_SUFFIX",
+    "truncate_rendered_value",
+]

shared/template_rendering/tests/template_rendering/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

shared/template_rendering/tests/template_rendering/test_truncate_rendered_value.py

@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow_shared.template_rendering import (
+    TRUNCATE_MIN_CONTENT_LENGTH,
+    TRUNCATE_PREFIX,
+    TRUNCATE_SUFFIX,
+    truncate_rendered_value,
+)
+
+
+def test_truncate_rendered_value_prioritizes_message():
+    """Test that truncation message is always shown first, content only if space allows."""
+    trunc_only = f"{TRUNCATE_PREFIX}{TRUNCATE_SUFFIX}"
+    trunc_only_len = len(trunc_only)
+    overhead = len(TRUNCATE_PREFIX) + len(TRUNCATE_SUFFIX)
+    min_length_for_content = overhead + TRUNCATE_MIN_CONTENT_LENGTH
+
+    test_cases = [
+        (1, "test", "Minimum value"),
+        (3, "test", "At ellipsis length"),
+        (5, "test", "Very small"),
+        (10, "password123", "Small"),
+        (20, "secret_value", "Small with content"),
+        (50, "This is a test string", "Medium"),
+        (overhead + 1, "Hello World", "At prefix+suffix boundary v1"),
+        (overhead + 2, "Hello World", "Just above boundary v1"),
+        (min_length_for_content - 3, "Hello World", "At overhead boundary v2"),
+        (90, "short", "Normal case - short string"),
+        (100, "This is a longer string", "Normal case"),
+        (
+            150,
+            "x" * 200,
+            "Large max_length, long string",
+        ),
+        (100, "None", "String 'None'"),
+        (100, "True", "String 'True'"),
+        (100, "{'key': 'value'}", "Dict-like string"),
+        (100, "test's", "String with apostrophe"),
+        (90, '"quoted"', "String with quotes"),
+    ]
+
+    for max_length, rendered, description in test_cases:
+        result = truncate_rendered_value(rendered, max_length)
+
+        assert result.startswith(TRUNCATE_PREFIX), (
+            f"Failed for {description}: result should start with prefix"
+        )
+
+        if max_length < trunc_only_len or max_length < min_length_for_content:
+            assert result == trunc_only, (
+                f"Failed for {description}: max_length={max_length}, expected message only, got: {result}"
+            )
+        else:
+            assert result.endswith(TRUNCATE_SUFFIX), (
+                f"Failed for {description}: result should end with suffix"
+            )
+            assert len(result) <= max_length, (
+                f"Failed for {description}: result length {len(result)} > max_length {max_length}"
+            )
+
+
+def test_truncate_rendered_value_exact_expected_output():
+    """Test that truncation produces exact expected output: message first, then content when space allows."""
+    trunc_only = TRUNCATE_PREFIX + TRUNCATE_SUFFIX
+    overhead = len(TRUNCATE_PREFIX) + len(TRUNCATE_SUFFIX)
+    min_length_for_content = overhead + TRUNCATE_MIN_CONTENT_LENGTH
+
+    test_cases = [
+        (1, "test", trunc_only),
+        (3, "test", trunc_only),
+        (5, "test", trunc_only),
+        (10, "password123", trunc_only),
+        (20, "secret_value", trunc_only),
+        (50, "This is a test string", trunc_only),
+        (overhead + 1, "Hello World", trunc_only),
+        (overhead + 2, "Hello World", trunc_only),
+        (min_length_for_content - 3, "Hello World", trunc_only),
+        (90, "short", TRUNCATE_PREFIX + "short" + TRUNCATE_SUFFIX),
+        (100, "This is a longer string", TRUNCATE_PREFIX + "This is a longer st" + TRUNCATE_SUFFIX),
+        (150, "x" * 200, TRUNCATE_PREFIX + "x" * 69 + TRUNCATE_SUFFIX),
+        (100, "None", TRUNCATE_PREFIX + "None" + TRUNCATE_SUFFIX),
+        (100, "True", TRUNCATE_PREFIX + "True" + TRUNCATE_SUFFIX),
+        (100, "{'key': 'value'}", TRUNCATE_PREFIX + "{'key': 'value'}" + TRUNCATE_SUFFIX),
+        (100, "test's", TRUNCATE_PREFIX + "test's" + TRUNCATE_SUFFIX),
+        (90, '"quoted"', TRUNCATE_PREFIX + '"quoted"' + TRUNCATE_SUFFIX),
+    ]
+
+    for max_length, rendered, expected in test_cases:
+        result = truncate_rendered_value(rendered, max_length)
+        assert result == expected, (
+            f"max_length={max_length}, rendered={rendered!r}:\n"
+            f"  expected: {expected!r}\n"
+            f"  got:      {result!r}"
+        )

task-sdk/pyproject.toml

@@ -136,6 +136,7 @@ path = "src/airflow/sdk/__init__.py"
 "../shared/listeners/src/airflow_shared/listeners" = "src/airflow/sdk/_shared/listeners"
 "../shared/plugins_manager/src/airflow_shared/plugins_manager" = "src/airflow/sdk/_shared/plugins_manager"
 "../shared/providers_discovery/src/airflow_shared/providers_discovery" = "src/airflow/sdk/_shared/providers_discovery"
+"../shared/template_rendering/src/airflow_shared/template_rendering" = "src/airflow/sdk/_shared/template_rendering"
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/airflow"]

task-sdk/src/airflow/sdk/_shared/template_rendering

@@ -0,0 +1 @@
+../../../../../shared/template_rendering/src/airflow_shared/template_rendering