Deprecate Asset.extra

Author: uranusjr

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

@@ -90,59 +90,6 @@ The identifier does not have to be absolute; it can be a scheme-less, relative U
 
 Non-absolute identifiers are considered plain strings that do not carry any semantic meanings to Airflow.
 
-Extra information on assets
-----------------------------
-
-If needed, you can include an additional dictionary in an asset using the ``extra`` parameter:
-
-.. code-block:: python
-
-    example_asset = Asset(
-        "s3://asset/example.csv",
-        extra={"team": "trainees"},
-    )
-
-This allows you to provide custom metadata about the asset, such as ownership information or the purpose of the file. The ``extra`` field does **NOT** affect the identity of an asset.
-Thus, maintaining the uniqueness of the ``extra`` value is the user responsibility. It suggested to have only one single set of ``extra`` value per asset.
-
-For example, in the following snippet, only one of the ``extra`` dictionaries will ultimately be stored, but it does guaranteed which one will be stored.
-
-.. code-block:: python
-
-     Asset("s3://asset/example.csv", extra={"d": "e"})
-     Asset("s3://asset/example.csv", extra={"f": "g"})
-
-This behavior also applies to dynamically generated assets created through ``AssetAlias``.
-In the example below, the final stored ``extra`` value is not guaranteed and it might vary based on Dag processor settings.
-
-.. code-block:: python
-
-    from airflow.sdk import AssetAlias
-
-
-    @dag(schedule=None)
-    def my_dag_1():
-
-        @task(outlets=[AssetAlias("my-task-outputs")])
-        def my_task_with_outlet_events(*, outlet_events):
-            outlet_events[AssetAlias("my-task-outputs")].add(
-                # Asset extra set as {"from": "asset alias"}
-                Asset("s3://bucket/my-task", extra={"from": "asset alias"})
-            )
-
-        my_task_with_outlet_events()
-
-
-    # Asset extra set as {"key": "value"}
-    @dag(schedule=Asset("s3://bucket/my-task", extra={"key": "value"}))
-    def my_dag_2(): ...
-
-
-    my_dag_1()
-    my_dag_2()
-
-    # It's not guaranteed which extra will be the one stored
-
 Security Warnings
 ----------------------------
 
@@ -193,8 +140,7 @@ Attaching extra information to an emitting asset event
 
 .. versionadded:: 2.10.0
 
-A task with an asset outlet can optionally attach extra information before it emits an asset event. This is different
-from `Extra information on assets`_. Extra information on an asset statically describes the entity pointed to by the asset URI; extra information on the *asset event* instead should be used to annotate the triggering data change, such as how many rows in the database are changed by the update, or the date range covered by it.
+A task with an asset outlet can optionally attach extra information before it emits an asset event.
 
 The easiest way to attach extra information to the asset event is by ``yield``-ing a ``Metadata`` object from a task:
 
@@ -225,6 +171,11 @@ There's minimal magic here---Airflow simply writes the yielded values to the exa
 
 .. note:: Asset event extra information can only contain JSON-serializable values (list and dict nesting is possible). This is due to the value being stored in the database.
 
+.. versionchanged:: 3.2.0
+
+  Assets may also contain an extra dict, which is static information distinct from event extras.
+  This was considered confusing, has been deprecated, and will be removed in Airflow 4.
+
 .. _fetching_information_from_previously_emitted_asset_events:
 
 Fetching information from previously emitted asset events

airflow-core/src/airflow/models/taskinstance.py

@@ -1418,7 +1418,6 @@ def register_asset_changes_in_db(
         outlet_events: list[dict[str, Any]],
         session: Session = NEW_SESSION,
     ) -> None:
-        print(task_outlets, outlet_events)
         from airflow.serialization.definitions.assets import (
             SerializedAsset,
             SerializedAssetNameRef,

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

@@ -95,7 +95,6 @@ def _decode_asset(var: dict[str, Any]):
         name=var["name"],
         uri=var["uri"],
         group=var["group"],
-        extra=var["extra"],
         watchers=[
             SerializedAssetWatcher(
                 name=watcher["name"],
@@ -106,6 +105,8 @@ def _decode_asset(var: dict[str, Any]):
             )
             for watcher in watchers
         ],
+        # TODO: Deprecated in SDK. Remove in Airflow 4.0.
+        extra=var.get("extra", {}),
     )
 
 

airflow-core/src/airflow/serialization/definitions/assets.py

@@ -29,6 +29,7 @@
 if TYPE_CHECKING:
     from collections.abc import Callable, Iterable, Iterator, MutableSequence
 
+    from pydantic import JsonValue
     from typing_extensions import Self
 
     from airflow.models.asset import AssetModel
@@ -118,9 +119,11 @@ class SerializedAsset(SerializedAssetBase):
     name: str
     uri: str
     group: str
-    extra: dict[str, Any]
     watchers: MutableSequence[SerializedAssetWatcher]
 
+    # TODO: Deprecated in SDK. Remove in Airflow 4.0.
+    extra: dict[str, JsonValue]
+
     def as_expression(self) -> Any:
         """
         Serialize the asset into its scheduling expression.

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

@@ -179,7 +179,9 @@ def encode_asset_like(a: BaseAsset | SerializedAssetBase) -> dict[str, Any]:
     d: dict[str, Any]
     match a:
         case Asset() | SerializedAsset():
-            d = {"__type": DAT.ASSET, "name": a.name, "uri": a.uri, "group": a.group, "extra": a.extra}
+            d = {"__type": DAT.ASSET, "name": a.name, "uri": a.uri, "group": a.group}
+            if a.extra:
+                d["extra"] = a.extra
             if a.watchers:
                 d["watchers"] = [{"name": w.name, "trigger": encode_trigger(w.trigger)} for w in a.watchers]
             return d

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

@@ -28,6 +28,7 @@
 
 import attrs
 
+from airflow.sdk.exceptions import RemovedInAirflow4Warning
 from airflow.sdk.providers_manager_runtime import ProvidersManagerTaskRuntime
 
 if TYPE_CHECKING:
@@ -271,7 +272,7 @@ class Asset(os.PathLike, BaseAsset):
         default=attrs.Factory(operator.attrgetter("asset_type"), takes_self=True),
         validator=[_validate_identifier],
     )
-    extra: dict[str, JsonValue] = attrs.field(
+    _extra: dict[str, JsonValue] = attrs.field(
         factory=dict,
         converter=_set_extra_default,
     )
@@ -289,7 +290,6 @@ def __init__(
         uri: str | ObjectStoragePath,
         *,
         group: str = ...,
-        extra: dict[str, JsonValue] | None = None,
         watchers: list[AssetWatcher] = ...,
     ) -> None:
         """Canonical; both name and uri are provided."""
@@ -300,7 +300,6 @@ def __init__(
         name: str,
         *,
         group: str = ...,
-        extra: dict[str, JsonValue] | None = None,
         watchers: list[AssetWatcher] = ...,
     ) -> None:
         """It's possible to only provide the name, either by keyword or as the only positional argument."""
@@ -311,7 +310,6 @@ def __init__(
         *,
         uri: str | ObjectStoragePath,
         group: str = ...,
-        extra: dict[str, JsonValue] | None = None,
         watchers: list[AssetWatcher] = ...,
     ) -> None:
         """It's possible to only provide the URI as a keyword argument."""
@@ -342,6 +340,11 @@ def __init__(
         if group is not None:
             kwargs["group"] = group
         if extra is not None:
+            warnings.warn(
+                "Asset.extra is deprecated and will be removed in the future.",
+                RemovedInAirflow4Warning,
+                stacklevel=2,
+            )
             kwargs["extra"] = extra
         if watchers is not None:
             kwargs["watchers"] = watchers
@@ -404,6 +407,20 @@ def normalized_uri(self) -> str | None:
         except ValueError:
             return None
 
+    @property
+    def extra(self) -> dict[str, JsonValue]:
+        # No warning here; a warning is shown when the value is set.
+        return self._extra
+
+    @extra.setter
+    def extra(self, value: dict[str, JsonValue]) -> None:
+        warnings.warn(
+            "Asset.extra is deprecated and will be removed in the future.",
+            RemovedInAirflow4Warning,
+            stacklevel=2,
+        )
+        self._extra = value
+
 
 class AssetRef(BaseAsset, AttrsInstance):
     """