feat(task_sdk): Move assets.metadata to task_sdk.definitions.asset

airflow/example_dags/example_outlet_event_extra.py

@@ -28,8 +28,7 @@
 from airflow.decorators import task
 from airflow.models.dag import DAG
 from airflow.providers.standard.operators.bash import BashOperator
-from airflow.sdk.definitions.asset import Asset
-from airflow.sdk.definitions.asset.metadata import Metadata
+from airflow.sdk.definitions.asset import Asset, Metadata
 
 ds = Asset("s3://output/1.txt")
 

airflow/utils/operator_helpers.py

@@ -251,7 +251,7 @@ def __init__(
     def run(self, *args, **kwargs) -> Any:
         import inspect
 
-        from airflow.sdk.definitions.asset.metadata import Metadata
+        from airflow.sdk.definitions.asset import Metadata
         from airflow.utils.types import NOTSET
 
         if not inspect.isgeneratorfunction(self.func):

docs/apache-airflow/authoring-and-scheduling/datasets.rst

@@ -249,7 +249,7 @@ The easiest way to attach extra information to the asset event is by ``yield``-i
 .. code-block:: python
 
     from airflow.sdk.definitions.asset import Asset
-    from airflow.sdk.definitions.asset.metadata import Metadata
+    from airflow.sdk.definitions.asset import Metadata
 
     example_s3_asset = Asset("s3://asset/example.csv")
 
@@ -452,7 +452,7 @@ The following example creates an asset event against the S3 URI ``f"s3://bucket/
 
 .. code-block:: python
 
-    from airflow.sdk.definitions.asset.metadata import Metadata
+    from airflow.sdk.definitions.asset import Metadata
 
 
     @task(outlets=[AssetAlias("my-task-outputs")])

task_sdk/src/airflow/sdk/definitions/asset.py

@@ -25,6 +25,7 @@
 from typing import TYPE_CHECKING, Any, Callable, ClassVar, cast, overload
 
 import attr
+import attrs
 from sqlalchemy import select
 
 from airflow.api_internal.internal_api_call import internal_api_call
@@ -528,3 +529,22 @@ def as_expression(self) -> Any:
         :meta private:
         """
         return {"all": [o.as_expression() for o in self.objects]}
+
+
+@attrs.define(init=False)
+class Metadata:
+    """Metadata to attach to an AssetEvent."""
+
+    uri: str
+    extra: dict[str, Any]
+    alias_name: str | None = None
+
+    def __init__(
+        self, target: str | Asset, extra: dict[str, Any], alias: AssetAlias | str | None = None
+    ) -> None:
+        self.uri = extract_event_key(target)
+        self.extra = extra
+        if isinstance(alias, AssetAlias):
+            self.alias_name = alias.name
+        else:
+            self.alias_name = alias

tests/models/test_taskinstance.py

@@ -2472,8 +2472,7 @@ def write(*, outlet_events):
 
     @pytest.mark.skip_if_database_isolation_mode  # Does not work in db isolation mode
     def test_outlet_asset_extra_yield(self, dag_maker, session):
-        from airflow.sdk.definitions.asset import Asset
-        from airflow.sdk.definitions.asset.metadata import Metadata
+        from airflow.sdk.definitions.asset import Asset, Metadata
 
         with dag_maker(schedule=None, session=session) as dag:
 
@@ -2645,8 +2644,7 @@ def producer(*, outlet_events):
 
     @pytest.mark.skip_if_database_isolation_mode  # Does not work in db isolation mode
     def test_outlet_asset_alias_through_metadata(self, dag_maker, session):
-        from airflow.sdk.definitions.asset import AssetAlias
-        from airflow.sdk.definitions.asset.metadata import Metadata
+        from airflow.sdk.definitions.asset import AssetAlias, Metadata
 
         asset_uri = "test_outlet_asset_alias_through_metadata_ds"
         asset_alias_name = "test_outlet_asset_alias_through_metadata_asset_alias"