Use name passed to @asset decorator when fetching the asset. (#56434)

Author: tirkarthi

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

@@ -66,7 +66,7 @@ def from_definition(cls, definition: AssetDefinition | MultiAssetDefinition) ->
             ],
             outlets=[v for _, v in definition.iter_assets()],
             python_callable=definition._function,
-            definition_name=definition._function.__name__,
+            definition_name=definition.name,
         )
 
     def _iter_kwargs(self, context: Mapping[str, Any]) -> Iterator[tuple[str, Any]]:
@@ -137,6 +137,7 @@ class MultiAssetDefinition(BaseAsset):
     :meta private:
     """
 
+    name: str
     _function: Callable
     _source: asset.multi
 
@@ -231,7 +232,7 @@ def __call__(self, f: Callable) -> MultiAssetDefinition:
                 raise ValueError("nested function not supported")
             if not self.outlets:
                 raise ValueError("no outlets provided")
-            return MultiAssetDefinition(function=f, source=self)
+            return MultiAssetDefinition(function=f, source=self, name=f.__name__)
 
     def __call__(self, f: Callable) -> AssetDefinition:
         if f.__name__ != f.__qualname__:

task-sdk/tests/task_sdk/definitions/test_asset_decorators.py

@@ -437,3 +437,26 @@ def test_determine_kwargs_defaults(
         assert mock_supervisor_comms.mock_calls == [
             mock.call.send(GetAssetByName(name="inlet_asset_1")),
         ]
+
+    def test_from_definition_custom_name(self, mock_supervisor_comms, func_fixer):
+        @func_fixer
+        def example_asset_func(self):
+            pass
+
+        definition = asset(schedule=None, name="custom_name")(example_asset_func)
+        op = _AssetMainOperator.from_definition(definition)
+        assert op.task_id == "example_asset_func"
+        assert op.python_callable == example_asset_func
+        assert op._definition_name == "custom_name"
+
+        mock_supervisor_comms.send.side_effect = [
+            AssetResult(name="custom_name", uri="s3://bucket/object1", group="Asset")
+        ]
+
+        assert op.determine_kwargs(context={}) == {
+            "self": Asset(name="custom_name", uri="s3://bucket/object1", group="Asset")
+        }
+
+        assert mock_supervisor_comms.mock_calls == [
+            mock.call.send(GetAssetByName(name="custom_name", uri="s3://bucket/object1", group="Asset"))
+        ]