Repository: apache/airflow (language: python)

Issue:
 New "airflow assets" CLI subcommand
### Description

[User Presentation Considerations](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=311627076#AIP75NewAssetCentricSyntax-UserPresentationConsiderations)

* [x] `airflow assets list` https://github.com/apache/airflow/pull/44387
* [x] `airflow assets detail` https://github.com/apache/airflow/pull/44445
* [x] `airflow assets --alias` to list asset aliases https://github.com/apache/airflow/pull/44595
* [x] `airflow assets detail --alias` to view an asset alias https://github.com/apache/airflow/pull/44595
* [x] `airflow assets materialize` https://github.com/apache/airflow/pull/44558
    * something like `airflow dags trigger`
    * steps
        1. get an asset
        2. find the tasks reference this asset as an outlet
            1. raise an error if there's not
            2. raise an error if there're more than one (as of now)
        3. trigger that DAG

### Use case/motivation

[Rationale](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=311626828#AIP73ExpandedDataAwareness-Rationale)

### Related issues

_No response_

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)



Discussion / hints:

Fix patch:
diff --git a/airflow/api_fastapi/core_api/datamodels/assets.py b/airflow/api_fastapi/core_api/datamodels/assets.py
index adc32c2e4808f..638ee1cba6e29 100644
--- a/airflow/api_fastapi/core_api/datamodels/assets.py
+++ b/airflow/api_fastapi/core_api/datamodels/assets.py
@@ -53,7 +53,9 @@ class AssetResponse(BaseModel):
     """Asset serializer for responses."""
 
     id: int
+    name: str
     uri: str
+    group: str
     extra: dict | None = None
     created_at: datetime
     updated_at: datetime
diff --git a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
index 46fd0382eb982..25ade1d917195 100644
--- a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
+++ b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
@@ -5737,9 +5737,15 @@ components:
         id:
           type: integer
           title: Id
+        name:
+          type: string
+          title: Name
         uri:
           type: string
           title: Uri
+        group:
+          type: string
+          title: Group
         extra:
           anyOf:
           - type: object
@@ -5771,7 +5777,9 @@ components:
       type: object
       required:
       - id
+      - name
       - uri
+      - group
       - created_at
       - updated_at
       - consuming_dags
diff --git a/airflow/cli/cli_config.py b/airflow/cli/cli_config.py
index 5611004f83dff..d03ebd312600e 100644
--- a/airflow/cli/cli_config.py
+++ b/airflow/cli/cli_config.py
@@ -974,6 +974,13 @@ def string_lower_type(val):
     default=("dag_id", "fileloc", "owners", "is_paused"),
 )
 
+ARG_ASSET_LIST_COLUMNS = Arg(
+    ("--columns",),
+    type=string_list_type,
+    help="List of columns to render. (default: ['name', 'uri', 'group', 'extra'])",
+    default=("name", "uri", "group", "extra"),
+)
+
 ALTERNATIVE_CONN_SPECS_ARGS = [
     ARG_CONN_TYPE,
     ARG_CONN_DESCRIPTION,
@@ -1009,6 +1016,14 @@ class GroupCommand(NamedTuple):
 
 CLICommand = Union[ActionCommand, GroupCommand]
 
+ASSETS_COMMANDS = (
+    ActionCommand(
+        name="list",
+        help="List assets",
+        func=lazy_load_command("airflow.cli.commands.asset_command.asset_list"),
+        args=(ARG_OUTPUT, ARG_VERBOSE, ARG_ASSET_LIST_COLUMNS),
+    ),
+)
 BACKFILL_COMMANDS = (
     ActionCommand(
         name="create",
@@ -1856,6 +1871,11 @@ class GroupCommand(NamedTuple):
         help="Manage tasks",
         subcommands=TASKS_COMMANDS,
     ),
+    GroupCommand(
+        name="assets",
+        help="Manage assets",
+        subcommands=ASSETS_COMMANDS,
+    ),
     GroupCommand(
         name="pools",
         help="Manage pools",
diff --git a/airflow/cli/commands/asset_command.py b/airflow/cli/commands/asset_command.py
new file mode 100644
index 0000000000000..a43fe409021ba
--- /dev/null
+++ b/airflow/cli/commands/asset_command.py
@@ -0,0 +1,53 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import logging
+import typing
+
+from sqlalchemy import select
+
+from airflow.api_fastapi.core_api.datamodels.assets import AssetResponse
+from airflow.cli.simple_table import AirflowConsole
+from airflow.models.asset import AssetModel
+from airflow.utils import cli as cli_utils
+from airflow.utils.session import NEW_SESSION, provide_session
+
+if typing.TYPE_CHECKING:
+    from typing import Any
+
+    from sqlalchemy.orm import Session
+
+log = logging.getLogger(__name__)
+
+
+@cli_utils.action_cli
+@provide_session
+def asset_list(args, *, session: Session = NEW_SESSION) -> None:
+    """Display assets in the command line."""
+    assets = session.scalars(select(AssetModel).order_by(AssetModel.name))
+
+    def detail_mapper(asset: AssetModel) -> dict[str, Any]:
+        model = AssetResponse.model_validate(asset)
+        return model.model_dump(include=args.columns)
+
+    AirflowConsole().print_as(
+        data=assets,
+        output=args.output,
+        mapper=detail_mapper,
+    )
diff --git a/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow/ui/openapi-gen/requests/schemas.gen.ts
index f657ba8d4e2d8..84c63a7d7bc34 100644
--- a/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -240,10 +240,18 @@ export const $AssetResponse = {
       type: "integer",
       title: "Id",
     },
+    name: {
+      type: "string",
+      title: "Name",
+    },
     uri: {
       type: "string",
       title: "Uri",
     },
+    group: {
+      type: "string",
+      title: "Group",
+    },
     extra: {
       anyOf: [
         {
@@ -290,7 +298,9 @@ export const $AssetResponse = {
   type: "object",
   required: [
     "id",
+    "name",
     "uri",
+    "group",
     "created_at",
     "updated_at",
     "consuming_dags",
diff --git a/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow/ui/openapi-gen/requests/types.gen.ts
index e15de90441bb0..1adcdc8eba8f4 100644
--- a/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -68,7 +68,9 @@ export type AssetEventResponse = {
  */
 export type AssetResponse = {
   id: number;
+  name: string;
   uri: string;
+  group: string;
   extra?: {
     [key: string]: unknown;
   } | null;


Test patch:
diff --git a/tests/api_fastapi/core_api/routes/public/test_assets.py b/tests/api_fastapi/core_api/routes/public/test_assets.py
index 8976af87fa13b..0e7db3910531d 100644
--- a/tests/api_fastapi/core_api/routes/public/test_assets.py
+++ b/tests/api_fastapi/core_api/routes/public/test_assets.py
@@ -48,7 +48,9 @@ def _create_assets(session, num: int = 2) -> None:
     assets = [
         AssetModel(
             id=i,
+            name=f"simple{i}",
             uri=f"s3://bucket/key/{i}",
+            group="asset",
             extra={"foo": "bar"},
             created_at=timezone.parse(default_time),
             updated_at=timezone.parse(default_time),
@@ -64,7 +66,9 @@ def _create_assets_with_sensitive_extra(session, num: int = 2) -> None:
     assets = [
         AssetModel(
             id=i,
+            name=f"sensitive{i}",
             uri=f"s3://bucket/key/{i}",
+            group="asset",
             extra={"password": "bar"},
             created_at=timezone.parse(default_time),
             updated_at=timezone.parse(default_time),
@@ -219,7 +223,9 @@ def test_should_respond_200(self, test_client, session):
             "assets": [
                 {
                     "id": 1,
+                    "name": "simple1",
                     "uri": "s3://bucket/key/1",
+                    "group": "asset",
                     "extra": {"foo": "bar"},
                     "created_at": tz_datetime_format,
                     "updated_at": tz_datetime_format,
@@ -229,7 +235,9 @@ def test_should_respond_200(self, test_client, session):
                 },
                 {
                     "id": 2,
+                    "name": "simple2",
                     "uri": "s3://bucket/key/2",
+                    "group": "asset",
                     "extra": {"foo": "bar"},
                     "created_at": tz_datetime_format,
                     "updated_at": tz_datetime_format,
@@ -566,7 +574,9 @@ def test_should_respond_200(self, test_client, url, session):
         assert response.status_code == 200
         assert response.json() == {
             "id": 1,
+            "name": "simple1",
             "uri": "s3://bucket/key/1",
+            "group": "asset",
             "extra": {"foo": "bar"},
             "created_at": tz_datetime_format,
             "updated_at": tz_datetime_format,
@@ -594,7 +604,9 @@ def test_should_mask_sensitive_extra(self, test_client, session):
         assert response.status_code == 200
         assert response.json() == {
             "id": 1,
+            "name": "sensitive1",
             "uri": "s3://bucket/key/1",
+            "group": "asset",
             "extra": {"password": "***"},
             "created_at": tz_datetime_format,
             "updated_at": tz_datetime_format,
diff --git a/tests/cli/commands/test_asset_command.py b/tests/cli/commands/test_asset_command.py
new file mode 100644
index 0000000000000..0cb4d8dbd8432
--- /dev/null
+++ b/tests/cli/commands/test_asset_command.py
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import typing
+
+import pytest
+
+from airflow.cli import cli_parser
+from airflow.cli.commands import asset_command
+from airflow.models.dagbag import DagBag
+
+from tests_common.test_utils.config import conf_vars
+from tests_common.test_utils.db import clear_db_dags
+
+if typing.TYPE_CHECKING:
+    from argparse import ArgumentParser
+
+pytestmark = [pytest.mark.db_test]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def prepare_examples():
+    DagBag(include_examples=True).sync_to_db()
+    yield
+    clear_db_dags()
+
+
+@pytest.fixture(scope="module")
+def parser() -> ArgumentParser:
+    return cli_parser.get_parser()
+
+
+@conf_vars({("core", "load_examples"): "true"})
+def test_cli_assets_list(parser: ArgumentParser) -> None:
+    args = parser.parse_args(["assets", "list", "--output=json"])
+    with contextlib.redirect_stdout(io.StringIO()) as temp_stdout:
+        asset_command.asset_list(args)
+
+    asset_list = json.loads(temp_stdout.getvalue())
+    assert len(asset_list) > 0
+    assert "name" in asset_list[0]
+    assert "uri" in asset_list[0]
+    assert "group" in asset_list[0]
+    assert "extra" in asset_list[0]
+    assert any(asset["uri"] == "s3://dag1/output_1.txt" for asset in asset_list)