Deprecate accessing inlet and outlet events through string

airflow/api_connexion/endpoints/asset_endpoint.py

@@ -45,7 +45,6 @@
 )
 from airflow.assets.manager import asset_manager
 from airflow.models.asset import AssetDagRunQueue, AssetEvent, AssetModel
-from airflow.sdk.definitions.asset import Asset
 from airflow.utils import timezone
 from airflow.utils.api_migration import mark_fastapi_migration_done
 from airflow.utils.db import get_query_count
@@ -341,15 +340,16 @@ def create_asset_event(session: Session = NEW_SESSION) -> APIResponse:
     except ValidationError as err:
         raise BadRequest(detail=str(err))
 
+    # TODO: handle name
     uri = json_body["asset_uri"]
-    asset = session.scalar(select(AssetModel).where(AssetModel.uri == uri).limit(1))
-    if not asset:
+    asset_model = session.scalar(select(AssetModel).where(AssetModel.uri == uri).limit(1))
+    if not asset_model:
         raise NotFound(title="Asset not found", detail=f"Asset with uri: '{uri}' not found")
     timestamp = timezone.utcnow()
     extra = json_body.get("extra", {})
     extra["from_rest_api"] = True
     asset_event = asset_manager.register_asset_change(
-        asset=Asset(uri=uri),
+        asset=asset_model.to_public(),
         timestamp=timestamp,
         extra=extra,
         session=session,

airflow/api_connexion/schemas/asset_schema.py

@@ -70,6 +70,7 @@ class Meta:
 
     id = auto_field()
     name = auto_field()
+    group = auto_field()
 
 
 class AssetSchema(SQLAlchemySchema):
@@ -82,6 +83,8 @@ class Meta:
 
     id = auto_field()
     uri = auto_field()
+    name = auto_field()
+    group = auto_field()
     extra = JsonObjectField()
     created_at = auto_field()
     updated_at = auto_field()

airflow/api_fastapi/core_api/datamodels/assets.py

@@ -47,6 +47,7 @@ class AssetAliasSchema(BaseModel):
 
     id: int
     name: str
+    group: str
 
 
 class AssetResponse(BaseModel):

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -5742,10 +5742,14 @@ components:
         name:
           type: string
           title: Name
+        group:
+          type: string
+          title: Group
       type: object
       required:
       - id
       - name
+      - group
       title: AssetAliasSchema
       description: Asset alias serializer for assets.
     AssetCollectionResponse:

airflow/api_fastapi/core_api/routes/public/assets.py

@@ -51,7 +51,6 @@
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.assets.manager import asset_manager
 from airflow.models.asset import AssetDagRunQueue, AssetEvent, AssetModel
-from airflow.sdk.definitions.asset import Asset
 from airflow.utils import timezone
 
 assets_router = AirflowRouter(tags=["Asset"])
@@ -171,13 +170,13 @@ def create_asset_event(
     session: SessionDep,
 ) -> AssetEventResponse:
     """Create asset events."""
-    asset = session.scalar(select(AssetModel).where(AssetModel.uri == body.uri).limit(1))
-    if not asset:
+    asset_model = session.scalar(select(AssetModel).where(AssetModel.uri == body.uri).limit(1))
+    if not asset_model:
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"Asset with uri: `{body.uri}` was not found")
     timestamp = timezone.utcnow()
 
     assets_event = asset_manager.register_asset_change(
-        asset=Asset(uri=body.uri),
+        asset=asset_model.to_public(),
         timestamp=timestamp,
         extra=body.extra,
         session=session,

airflow/assets/manager.py

@@ -86,16 +86,16 @@ def _add_one(asset_alias: AssetAlias) -> AssetAliasModel:
     def _add_asset_alias_association(
         cls,
         alias_names: Collection[str],
-        asset: AssetModel,
+        asset_model: AssetModel,
         *,
         session: Session,
     ) -> None:
-        already_related = {m.name for m in asset.aliases}
+        already_related = {m.name for m in asset_model.aliases}
         existing_aliases = {
             m.name: m
             for m in session.scalars(select(AssetAliasModel).where(AssetAliasModel.name.in_(alias_names)))
         }
-        asset.aliases.extend(
+        asset_model.aliases.extend(
             existing_aliases.get(name, AssetAliasModel(name=name))
             for name in alias_names
             if name not in already_related
@@ -121,7 +121,7 @@ def register_asset_change(
         """
         asset_model = session.scalar(
             select(AssetModel)
-            .where(AssetModel.uri == asset.uri)
+            .where(AssetModel.name == asset.name, AssetModel.uri == asset.uri)
             .options(
                 joinedload(AssetModel.aliases),
                 joinedload(AssetModel.consuming_dags).joinedload(DagScheduleAssetReference.dag),
@@ -131,7 +131,9 @@ def register_asset_change(
             cls.logger().warning("AssetModel %s not found", asset)
             return None
 
-        cls._add_asset_alias_association({alias.name for alias in aliases}, asset_model, session=session)
+        cls._add_asset_alias_association(
+            alias_names={alias.name for alias in aliases}, asset_model=asset_model, session=session
+        )
 
         event_kwargs = {
             "asset_id": asset_model.id,

airflow/example_dags/example_asset_alias.py

@@ -67,7 +67,7 @@ def produce_asset_events():
     def produce_asset_events_through_asset_alias(*, outlet_events=None):
         bucket_name = "bucket"
         object_path = "my-task"
-        outlet_events["example-alias"].add(Asset(f"s3://{bucket_name}/{object_path}"))
+        outlet_events[AssetAlias("example-alias")].add(Asset(f"s3://{bucket_name}/{object_path}"))
 
     produce_asset_events_through_asset_alias()
 

airflow/example_dags/example_asset_alias_with_no_taskflow.py

@@ -68,7 +68,7 @@ def produce_asset_events():
     def produce_asset_events_through_asset_alias_with_no_taskflow(*, outlet_events=None):
         bucket_name = "bucket"
         object_path = "my-task"
-        outlet_events["example-alias-no-taskflow"].add(Asset(f"s3://{bucket_name}/{object_path}"))
+        outlet_events[AssetAlias("example-alias-no-taskflow")].add(Asset(f"s3://{bucket_name}/{object_path}"))
 
     PythonOperator(
         task_id="produce_asset_events_through_asset_alias_with_no_taskflow",

airflow/lineage/hook.py

@@ -95,24 +95,40 @@ def _generate_key(self, asset: Asset, context: LineageContext) -> str:
         return f"{asset.uri}_{extra_hash}_{id(context)}"
 
     def create_asset(
-        self, scheme: str | None, uri: str | None, asset_kwargs: dict | None, asset_extra: dict | None
+        self,
+        *,
+        scheme: str | None = None,
+        uri: str | None = None,
+        name: str | None = None,
+        group: str | None = None,
+        asset_kwargs: dict | None = None,
+        asset_extra: dict | None = None,
     ) -> Asset | None:
         """
         Create an asset instance using the provided parameters.
 
         This method attempts to create an asset instance using the given parameters.
-        It first checks if a URI is provided and falls back to using the default asset factory
-        with the given URI if no other information is available.
+        It first checks if a URI or a name is provided and falls back to using the default asset factory
+        with the given URI or name if no other information is available.
 
-        If a scheme is provided but no URI, it attempts to find an asset factory that matches
+        If a scheme is provided but no URI or name, it attempts to find an asset factory that matches
         the given scheme. If no such factory is found, it logs an error message and returns None.
 
         If asset_kwargs is provided, it is used to pass additional parameters to the asset
         factory. The asset_extra parameter is also passed to the factory as an ``extra`` parameter.
         """
-        if uri:
+        if uri or name:
             # Fallback to default factory using the provided URI
-            return Asset(uri=uri, extra=asset_extra)
+            kwargs: dict[str, str | dict] = {}
+            if uri:
+                kwargs["uri"] = uri
+            if name:
+                kwargs["name"] = name
+            if group:
+                kwargs["group"] = group
+            if asset_extra:
+                kwargs["extra"] = asset_extra
+            return Asset(**kwargs)  # type: ignore[call-overload]
 
         if not scheme:
             self.log.debug(
@@ -137,11 +153,15 @@ def add_input_asset(
         context: LineageContext,
         scheme: str | None = None,
         uri: str | None = None,
+        name: str | None = None,
+        group: str | None = None,
         asset_kwargs: dict | None = None,
         asset_extra: dict | None = None,
     ):
         """Add the input asset and its corresponding hook execution context to the collector."""
-        asset = self.create_asset(scheme=scheme, uri=uri, asset_kwargs=asset_kwargs, asset_extra=asset_extra)
+        asset = self.create_asset(
+            scheme=scheme, uri=uri, name=name, group=group, asset_kwargs=asset_kwargs, asset_extra=asset_extra
+        )
         if asset:
             key = self._generate_key(asset, context)
             if key not in self._inputs:
@@ -153,11 +173,15 @@ def add_output_asset(
         context: LineageContext,
         scheme: str | None = None,
         uri: str | None = None,
+        name: str | None = None,
+        group: str | None = None,
         asset_kwargs: dict | None = None,
         asset_extra: dict | None = None,
     ):
         """Add the output asset and its corresponding hook execution context to the collector."""
-        asset = self.create_asset(scheme=scheme, uri=uri, asset_kwargs=asset_kwargs, asset_extra=asset_extra)
+        asset = self.create_asset(
+            scheme=scheme, uri=uri, name=name, group=group, asset_kwargs=asset_kwargs, asset_extra=asset_extra
+        )
         if asset:
             key = self._generate_key(asset, context)
             if key not in self._outputs:

airflow/serialization/serialized_objects.py

@@ -44,7 +44,11 @@
 from airflow.models.connection import Connection
 from airflow.models.dag import DAG, DagModel
 from airflow.models.dagrun import DagRun
-from airflow.models.expandinput import EXPAND_INPUT_EMPTY, create_expand_input, get_map_type_key
+from airflow.models.expandinput import (
+    EXPAND_INPUT_EMPTY,
+    create_expand_input,
+    get_map_type_key,
+)
 from airflow.models.mappedoperator import MappedOperator
 from airflow.models.param import Param, ParamsDict
 from airflow.models.taskinstance import SimpleTaskInstance, TaskInstance
@@ -213,7 +217,9 @@ def _get_registered_timetable(importable_string: str) -> type[Timetable] | None:
         return None
 
 
-def _get_registered_priority_weight_strategy(importable_string: str) -> type[PriorityWeightStrategy] | None:
+def _get_registered_priority_weight_strategy(
+    importable_string: str,
+) -> type[PriorityWeightStrategy] | None:
     from airflow import plugins_manager
 
     if importable_string in airflow_priority_weight_strategies:
@@ -256,13 +262,25 @@ def encode_asset_condition(var: BaseAsset) -> dict[str, Any]:
     :meta private:
     """
     if isinstance(var, Asset):
-        return {"__type": DAT.ASSET, "name": var.name, "uri": var.uri, "extra": var.extra}
+        return {
+            "__type": DAT.ASSET,
+            "name": var.name,
+            "uri": var.uri,
+            "group": var.group,
+            "extra": var.extra,
+        }
     if isinstance(var, AssetAlias):
-        return {"__type": DAT.ASSET_ALIAS, "name": var.name}
+        return {"__type": DAT.ASSET_ALIAS, "name": var.name, "group": var.group}
     if isinstance(var, AssetAll):
-        return {"__type": DAT.ASSET_ALL, "objects": [encode_asset_condition(x) for x in var.objects]}
+        return {
+            "__type": DAT.ASSET_ALL,
+            "objects": [encode_asset_condition(x) for x in var.objects],
+        }
     if isinstance(var, AssetAny):
-        return {"__type": DAT.ASSET_ANY, "objects": [encode_asset_condition(x) for x in var.objects]}
+        return {
+            "__type": DAT.ASSET_ANY,
+            "objects": [encode_asset_condition(x) for x in var.objects],
+        }
     raise ValueError(f"serialization not implemented for {type(var).__name__!r}")
 
 
@@ -274,22 +292,22 @@ def decode_asset_condition(var: dict[str, Any]) -> BaseAsset:
     """
     dat = var["__type"]
     if dat == DAT.ASSET:
-        return Asset(uri=var["uri"], name=var["name"], extra=var["extra"])
+        return Asset(name=var["name"], uri=var["uri"], group=var["group"], extra=var["extra"])
     if dat == DAT.ASSET_ALL:
         return AssetAll(*(decode_asset_condition(x) for x in var["objects"]))
     if dat == DAT.ASSET_ANY:
         return AssetAny(*(decode_asset_condition(x) for x in var["objects"]))
     if dat == DAT.ASSET_ALIAS:
-        return AssetAlias(name=var["name"])
+        return AssetAlias(name=var["name"], group=var["group"])
     raise ValueError(f"deserialization not implemented for DAT {dat!r}")
 
 
 def encode_outlet_event_accessor(var: OutletEventAccessor) -> dict[str, Any]:
-    raw_key = var.raw_key
+    key = var.key
     return {
         "extra": var.extra,
         "asset_alias_events": var.asset_alias_events,
-        "raw_key": BaseSerialization.serialize(raw_key),
+        "key": BaseSerialization.serialize(key),
     }
 
 
@@ -298,7 +316,7 @@ def decode_outlet_event_accessor(var: dict[str, Any]) -> OutletEventAccessor:
 
     outlet_event_accessor = OutletEventAccessor(
         extra=var["extra"],
-        raw_key=BaseSerialization.deserialize(var["raw_key"]),
+        key=BaseSerialization.deserialize(var["key"]),
         asset_alias_events=asset_alias_events,
     )
     return outlet_event_accessor
@@ -586,7 +604,9 @@ def _is_excluded(cls, var: Any, attrname: str, instance: Any) -> bool:
 
     @classmethod
     def serialize_to_json(
-        cls, object_to_serialize: BaseOperator | MappedOperator | DAG, decorated_fields: set
+        cls,
+        object_to_serialize: BaseOperator | MappedOperator | DAG,
+        decorated_fields: set,
     ) -> dict[str, Any]:
         """Serialize an object to JSON."""
         serialized_object: dict[str, Any] = {}
@@ -653,7 +673,11 @@ def serialize(
             return cls._encode(json_pod, type_=DAT.POD)
         elif isinstance(var, OutletEventAccessors):
             return cls._encode(
-                cls.serialize(var._dict, strict=strict, use_pydantic_models=use_pydantic_models),  # type: ignore[attr-defined]
+                cls.serialize(
+                    var._dict,  # type: ignore[attr-defined]
+                    strict=strict,
+                    use_pydantic_models=use_pydantic_models,
+                ),
                 type_=DAT.ASSET_EVENT_ACCESSORS,
             )
         elif isinstance(var, OutletEventAccessor):
@@ -696,15 +720,23 @@ def serialize(
         elif isinstance(var, (KeyError, AttributeError)):
             return cls._encode(
                 cls.serialize(
-                    {"exc_cls_name": var.__class__.__name__, "args": [var.args], "kwargs": {}},
+                    {
+                        "exc_cls_name": var.__class__.__name__,
+                        "args": [var.args],
+                        "kwargs": {},
+                    },
                     use_pydantic_models=use_pydantic_models,
                     strict=strict,
                 ),
                 type_=DAT.BASE_EXC_SER,
             )
         elif isinstance(var, BaseTrigger):
             return cls._encode(
-                cls.serialize(var.serialize(), use_pydantic_models=use_pydantic_models, strict=strict),
+                cls.serialize(
+                    var.serialize(),
+                    use_pydantic_models=use_pydantic_models,
+                    strict=strict,
+                ),
                 type_=DAT.BASE_TRIGGER,
             )
         elif callable(var):
@@ -1065,11 +1097,11 @@ def detect_task_dependencies(task: Operator) -> list[DagDependency]:
                         source=task.dag_id,
                         target="asset",
                         dependency_type="asset",
-                        dependency_id=obj.uri,
+                        dependency_id=obj.name,
                     )
                 )
             elif isinstance(obj, AssetAlias):
-                cond = AssetAliasCondition(obj.name)
+                cond = AssetAliasCondition(name=obj.name, group=obj.group)
 
                 deps.extend(cond.iter_dag_dependencies(source=task.dag_id, target=""))
         return deps
@@ -1298,7 +1330,11 @@ def populate_operator(cls, op: Operator, encoded_op: dict[str, Any]) -> None:
             # The case for "If OperatorLinks are defined in the operator that is being Serialized"
             # is handled in the deserialization loop where it matches k == "_operator_extra_links"
             if op_extra_links_from_plugin and "_operator_extra_links" not in encoded_op:
-                setattr(op, "operator_extra_links", list(op_extra_links_from_plugin.values()))
+                setattr(
+                    op,
+                    "operator_extra_links",
+                    list(op_extra_links_from_plugin.values()),
+                )
 
         for k, v in encoded_op.items():
             # python_callable_name only serves to detect function name changes