Revert "Support templated extra in outlets assets (#54885)" (#55199)

This reverts commit 438b76c.

Author: uranusjr

airflow-core/docs/authoring-and-scheduling/assets.rst

@@ -104,27 +104,6 @@ If needed, you can include an extra dictionary in an asset:
 
 This can be used to supply custom description to the asset, such as who has ownership to the target file, or what the file is for. The extra information does not affect an asset's identity.
 
-You can also use Jinja templating in the extra dictionary to enrich the asset with runtime information, such as the execution date of the task that emits events of the asset:
-
-.. code-block::
-
-    BashOperator(
-        task_id="write_example_asset",
-        bash_command="echo 'writing...'",
-        outlets=Asset(
-            "asset_example",
-            extra={
-                "static_extra": "value",
-                "dag_id": "{{ dag.dag_id }}",
-                "nested_extra": {
-                    "run_id": "{{ run_id }}",
-                    "logical_date": "{{ ds }}",
-                }
-            }
-        ),
-    )
-
-
 .. note:: **Security Note:** Asset URI and extra fields are not encrypted, they are stored in cleartext in Airflow's metadata database. Do NOT store any sensitive values, especially credentials, in either asset URIs or extra key values!
 
 Creating a task to emit asset events

task-sdk/src/airflow/sdk/definitions/_internal/templater.py

@@ -89,7 +89,7 @@ def prepare_template(self) -> None:
 
     def resolve_template_files(self) -> None:
         """Get the content of files for template_field / template_ext."""
-        if getattr(self, "template_ext", None):
+        if self.template_ext:
             for field in self.template_fields:
                 content = getattr(self, field, None)
                 if isinstance(content, str) and content.endswith(tuple(self.template_ext)):
@@ -170,7 +170,7 @@ def render_template(
             jinja_env = self.get_template_env()
 
         if isinstance(value, str):
-            if hasattr(self, "template_ext") and value.endswith(tuple(self.template_ext)):  # A filepath.
+            if value.endswith(tuple(self.template_ext)):  # A filepath.
                 template = jinja_env.get_template(value)
             else:
                 template = jinja_env.from_string(value)

task-sdk/src/airflow/sdk/definitions/asset/__init__.py

@@ -29,17 +29,13 @@
 import attrs
 
 from airflow.sdk.api.datamodels._generated import AssetProfile
-from airflow.sdk.definitions._internal.templater import Templater
 from airflow.serialization.dag_dependency import DagDependency
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Iterator
     from urllib.parse import SplitResult
 
-    import jinja2
-
     from airflow.models.asset import AssetModel
-    from airflow.sdk import Context
     from airflow.sdk.io.path import ObjectStoragePath
     from airflow.serialization.serialized_objects import SerializedAssetWatcher
     from airflow.triggers.base import BaseEventTrigger
@@ -309,7 +305,7 @@ def __init__(
 
 
 @attrs.define(init=False, unsafe_hash=False)
-class Asset(os.PathLike, BaseAsset, Templater):
+class Asset(os.PathLike, BaseAsset):
     """A representation of data asset dependencies between workflows."""
 
     name: str = attrs.field(
@@ -493,22 +489,6 @@ def asprofile(self) -> AssetProfile:
         """
         return AssetProfile(name=self.name or None, uri=self.uri or None, type=Asset.__name__)
 
-    def render_extra_field(
-        self,
-        context: Context,
-        jinja_env: jinja2.Environment | None = None,
-    ) -> None:
-        """
-        Template extra attribute.
-
-        :param context: Context dict with values to apply on content.
-        :param jinja_env: Jinja environment to use for rendering.
-        """
-        dag = context["dag"]
-        if not jinja_env:
-            jinja_env = self.get_template_env(dag=dag)
-        self._do_render_template_fields(self, ("extra",), context, jinja_env, set())
-
 
 class AssetRef(BaseAsset, AttrsInstance):
     """

task-sdk/src/airflow/sdk/execution_time/task_runner.py

@@ -1287,11 +1287,6 @@ def _execute_task(context: Context, ti: RuntimeTaskInstance, log: Logger):
 
     outlet_events = context_get_outlet_events(context)
 
-    for outlet in task.outlets or ():
-        if isinstance(outlet, Asset):
-            outlet.render_extra_field(context, jinja_env=task.dag.get_template_env())
-            outlet_events[outlet].extra.update(outlet.extra)
-
     if (pre_execute_hook := task._pre_execute_hook) is not None:
         create_executable_runner(pre_execute_hook, outlet_events, logger=log).run(context)
     if getattr(pre_execute_hook := task.pre_execute, "__func__", None) is not BaseOperator.pre_execute:

task-sdk/tests/task_sdk/execution_time/test_task_runner.py

@@ -942,38 +942,10 @@ def test_dag_parsing_context(make_ti_context, mock_supervisor_comms, monkeypatch
                 task_outlets=[
                     AssetProfile(name="s3://bucket/my-task", uri="s3://bucket/my-task", type="Asset")
                 ],
-                outlet_events=[
-                    {
-                        "dest_asset_key": {"name": "s3://bucket/my-task", "uri": "s3://bucket/my-task"},
-                        "extra": {},
-                    }
-                ],
+                outlet_events=[],
             ),
             id="asset",
         ),
-        pytest.param(
-            [
-                Asset(
-                    name="s3://bucket/my-task",
-                    uri="s3://bucket/my-task",
-                    extra={"task_id": "{{ task.task_id }}"},
-                )
-            ],
-            SucceedTask(
-                state="success",
-                end_date=timezone.datetime(2024, 12, 3, 10, 0),
-                task_outlets=[
-                    AssetProfile(name="s3://bucket/my-task", uri="s3://bucket/my-task", type="Asset")
-                ],
-                outlet_events=[
-                    {
-                        "dest_asset_key": {"name": "s3://bucket/my-task", "uri": "s3://bucket/my-task"},
-                        "extra": {"task_id": "asset-outlet-task"},
-                    }
-                ],
-            ),
-            id="asset_with_template_extra",
-        ),
         pytest.param(
             [Dataset(name="s3://bucket/my-task", uri="s3://bucket/my-task")],
             SucceedTask(
@@ -982,38 +954,10 @@ def test_dag_parsing_context(make_ti_context, mock_supervisor_comms, monkeypatch
                 task_outlets=[
                     AssetProfile(name="s3://bucket/my-task", uri="s3://bucket/my-task", type="Asset")
                 ],
-                outlet_events=[
-                    {
-                        "dest_asset_key": {"name": "s3://bucket/my-task", "uri": "s3://bucket/my-task"},
-                        "extra": {},
-                    }
-                ],
+                outlet_events=[],
             ),
             id="dataset",
         ),
-        pytest.param(
-            [
-                Dataset(
-                    name="s3://bucket/my-task",
-                    uri="s3://bucket/my-task",
-                    extra={"task_id": "{{ task.task_id }}"},
-                )
-            ],
-            SucceedTask(
-                state="success",
-                end_date=timezone.datetime(2024, 12, 3, 10, 0),
-                task_outlets=[
-                    AssetProfile(name="s3://bucket/my-task", uri="s3://bucket/my-task", type="Asset")
-                ],
-                outlet_events=[
-                    {
-                        "dest_asset_key": {"name": "s3://bucket/my-task", "uri": "s3://bucket/my-task"},
-                        "extra": {"task_id": "asset-outlet-task"},
-                    }
-                ],
-            ),
-            id="dataset_with_template_extra",
-        ),
         pytest.param(
             [Model(name="s3://bucket/my-task", uri="s3://bucket/my-task")],
             SucceedTask(
@@ -1022,38 +966,10 @@ def test_dag_parsing_context(make_ti_context, mock_supervisor_comms, monkeypatch
                 task_outlets=[
                     AssetProfile(name="s3://bucket/my-task", uri="s3://bucket/my-task", type="Asset")
                 ],
-                outlet_events=[
-                    {
-                        "dest_asset_key": {"name": "s3://bucket/my-task", "uri": "s3://bucket/my-task"},
-                        "extra": {},
-                    }
-                ],
+                outlet_events=[],
             ),
             id="model",
         ),
-        pytest.param(
-            [
-                Model(
-                    name="s3://bucket/my-task",
-                    uri="s3://bucket/my-task",
-                    extra={"task_id": "{{ task.task_id }}"},
-                )
-            ],
-            SucceedTask(
-                state="success",
-                end_date=timezone.datetime(2024, 12, 3, 10, 0),
-                task_outlets=[
-                    AssetProfile(name="s3://bucket/my-task", uri="s3://bucket/my-task", type="Asset")
-                ],
-                outlet_events=[
-                    {
-                        "dest_asset_key": {"name": "s3://bucket/my-task", "uri": "s3://bucket/my-task"},
-                        "extra": {"task_id": "asset-outlet-task"},
-                    }
-                ],
-            ),
-            id="model_with_template_extra",
-        ),
         pytest.param(
             [Asset.ref(name="s3://bucket/my-task")],
             SucceedTask(