(airflow) branch main updated: Add 'airflow assets details' subcommand (#44445)

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

weilee pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 79a9c828698 Add 'airflow assets details' subcommand (#44445)
79a9c828698 is described below

commit 79a9c828698c0eba9908ff8d3d6073bfa040494f
Author: Tzu-ping Chung <ur...@gmail.com>
AuthorDate: Fri Nov 29 10:05:29 2024 +0800

    Add 'airflow assets details' subcommand (#44445)
    
    * Add 'airflow assets details' subcommand
    
    * Limit unnecessary assets fetched
---
 airflow/cli/cli_config.py                |  9 +++++++++
 airflow/cli/commands/asset_command.py    | 32 ++++++++++++++++++++++++++++++++
 tests/cli/commands/test_asset_command.py | 26 ++++++++++++++++++++++++++
 3 files changed, 67 insertions(+)

diff --git a/airflow/cli/cli_config.py b/airflow/cli/cli_config.py
index 0c9703ec780..1d113a00ed9 100644
--- a/airflow/cli/cli_config.py
+++ b/airflow/cli/cli_config.py
@@ -935,6 +935,9 @@ ARG_ASSET_LIST_COLUMNS = Arg(
     default=("name", "uri", "group", "extra"),
 )
 
+ARG_ASSET_NAME = Arg(("--name",), help="Asset name")
+ARG_ASSET_URI = Arg(("--uri",), help="Asset URI")
+
 ALTERNATIVE_CONN_SPECS_ARGS = [
     ARG_CONN_TYPE,
     ARG_CONN_DESCRIPTION,
@@ -977,6 +980,12 @@ ASSETS_COMMANDS = (
         func=lazy_load_command("airflow.cli.commands.asset_command.asset_list"),
         args=(ARG_OUTPUT, ARG_VERBOSE, ARG_ASSET_LIST_COLUMNS),
     ),
+    ActionCommand(
+        name="details",
+        help="Show asset details",
+        func=lazy_load_command("airflow.cli.commands.asset_command.asset_details"),
+        args=(ARG_ASSET_NAME, ARG_ASSET_URI, ARG_OUTPUT, ARG_VERBOSE),
+    ),
 )
 BACKFILL_COMMANDS = (
     ActionCommand(
diff --git a/airflow/cli/commands/asset_command.py b/airflow/cli/commands/asset_command.py
index a43fe409021..5a8e67b5845 100644
--- a/airflow/cli/commands/asset_command.py
+++ b/airflow/cli/commands/asset_command.py
@@ -51,3 +51,35 @@ def asset_list(args, *, session: Session = NEW_SESSION) -> None:
         output=args.output,
         mapper=detail_mapper,
     )
+
+
+@cli_utils.action_cli
+@provide_session
+def asset_details(args, *, session: Session = NEW_SESSION) -> None:
+    """Display details of an asset."""
+    if not args.name and not args.uri:
+        raise SystemExit("Either --name or --uri is required")
+
+    stmt = select(AssetModel)
+    select_message_parts = []
+    if args.name:
+        stmt = stmt.where(AssetModel.name == args.name)
+        select_message_parts.append(f"name {args.name}")
+    if args.uri:
+        stmt = stmt.where(AssetModel.uri == args.uri)
+        select_message_parts.append(f"URI {args.uri}")
+    asset_it = iter(session.scalars(stmt.limit(2)))
+    select_message = " and ".join(select_message_parts)
+
+    if (asset := next(asset_it, None)) is None:
+        raise SystemExit(f"Asset with {select_message} does not exist.")
+    if next(asset_it, None) is not None:
+        raise SystemExit(f"More than one asset exists with {select_message}.")
+
+    model_data = AssetResponse.model_validate(asset).model_dump()
+    if args.output in ["table", "plain"]:
+        data = [{"property_name": key, "property_value": value} for key, value in model_data.items()]
+    else:
+        data = [model_data]
+
+    AirflowConsole().print_as(data=data, output=args.output)
diff --git a/tests/cli/commands/test_asset_command.py b/tests/cli/commands/test_asset_command.py
index 0cb4d8dbd84..cc5a4f1cda8 100644
--- a/tests/cli/commands/test_asset_command.py
+++ b/tests/cli/commands/test_asset_command.py
@@ -63,3 +63,29 @@ def test_cli_assets_list(parser: ArgumentParser) -> None:
     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)
+
+
+def test_cli_assets_details(parser: ArgumentParser) -> None:
+    args = parser.parse_args(["assets", "details", "--name=asset1_producer", "--output=json"])
+    with contextlib.redirect_stdout(io.StringIO()) as temp_stdout:
+        asset_command.asset_details(args)
+
+    asset_detail_list = json.loads(temp_stdout.getvalue())
+    assert len(asset_detail_list) == 1
+
+    # No good way to statically compare these.
+    undeterministic = {
+        "id": None,
+        "created_at": None,
+        "updated_at": None,
+        "consuming_dags": None,
+        "producing_tasks": None,
+    }
+
+    assert asset_detail_list[0] | undeterministic == undeterministic | {
+        "name": "asset1_producer",
+        "uri": "s3://bucket/asset1_producer",
+        "group": "asset",
+        "extra": {},
+        "aliases": [],
+    }