[airflow] 05/08: Add command to get DAG Details via CLI (#30432)

[ep...@apache.org]

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

ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit 06e4672bf44376b1dc146bc6cc37917460448ede
Author: maahir22 <56...@users.noreply.github.com>
AuthorDate: Fri Apr 14 22:44:48 2023 +0530

    Add command to get DAG Details via CLI (#30432)
    
    ---------
    
    Co-authored-by: Hussein Awala <hu...@awala.fr>
    Co-authored-by: Hussein Awala <ho...@gmail.com>
    (cherry picked from commit 7074167d71c93b69361d24c1121adc7419367f2a)
---
 airflow/cli/cli_config.py              |  6 ++++++
 airflow/cli/commands/dag_command.py    | 22 ++++++++++++++++++++++
 tests/cli/commands/test_dag_command.py | 20 ++++++++++++++++++++
 3 files changed, 48 insertions(+)

diff --git a/airflow/cli/cli_config.py b/airflow/cli/cli_config.py
index e08b4e01ec..ffcfd92de0 100644
--- a/airflow/cli/cli_config.py
+++ b/airflow/cli/cli_config.py
@@ -1101,6 +1101,12 @@ class GroupCommand(NamedTuple):
 CLICommand = Union[ActionCommand, GroupCommand]
 
 DAGS_COMMANDS = (
+    ActionCommand(
+        name="details",
+        help="Get DAG details given a DAG id",
+        func=lazy_load_command("airflow.cli.commands.dag_command.dag_details"),
+        args=(ARG_DAG_ID, ARG_OUTPUT, ARG_VERBOSE),
+    ),
     ActionCommand(
         name="list",
         help="List all the DAGs",
diff --git a/airflow/cli/commands/dag_command.py b/airflow/cli/commands/dag_command.py
index 8735a0e522..c81f1c0a47 100644
--- a/airflow/cli/commands/dag_command.py
+++ b/airflow/cli/commands/dag_command.py
@@ -32,6 +32,7 @@ from sqlalchemy.orm import Session
 
 from airflow import settings
 from airflow.api.client import get_current_api_client
+from airflow.api_connexion.schemas.dag_schema import dag_schema
 from airflow.cli.simple_table import AirflowConsole
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, RemovedInAirflow3Warning
@@ -346,6 +347,27 @@ def dag_list_dags(args) -> None:
     )
 
 
+@cli_utils.action_cli
+@suppress_logs_and_warning
+@provide_session
+def dag_details(args, session=NEW_SESSION):
+    """Get DAG details given a DAG id"""
+    dag = DagModel.get_dagmodel(args.dag_id, session=session)
+    if not dag:
+        raise SystemExit(f"DAG: {args.dag_id} does not exist in 'dag' table")
+    dag_detail = dag_schema.dump(dag)
+
+    if args.output in ["table", "plain"]:
+        data = [{"property_name": key, "property_value": value} for key, value in dag_detail.items()]
+    else:
+        data = [dag_detail]
+
+    AirflowConsole().print_as(
+        data=data,
+        output=args.output,
+    )
+
+
 @cli_utils.action_cli
 @suppress_logs_and_warning
 def dag_list_import_errors(args) -> None:
diff --git a/tests/cli/commands/test_dag_command.py b/tests/cli/commands/test_dag_command.py
index 6b4e2187fa..a54d26a993 100644
--- a/tests/cli/commands/test_dag_command.py
+++ b/tests/cli/commands/test_dag_command.py
@@ -31,6 +31,7 @@ import pytest
 import time_machine
 
 from airflow import settings
+from airflow.api_connexion.schemas.dag_schema import DAGSchema
 from airflow.cli import cli_parser
 from airflow.cli.commands import dag_command
 from airflow.exceptions import AirflowException
@@ -470,6 +471,25 @@ class TestCliDags:
         assert "airflow/example_dags/example_complex.py" in out
         assert "example_complex" in out
 
+    @conf_vars({("core", "load_examples"): "true"})
+    def test_cli_get_dag_details(self):
+        args = self.parser.parse_args(["dags", "details", "example_complex", "--output", "yaml"])
+        with contextlib.redirect_stdout(io.StringIO()) as temp_stdout:
+            dag_command.dag_details(args)
+            out = temp_stdout.getvalue()
+
+        dag_detail_fields = DAGSchema().fields.keys()
+
+        # Check if DAG Details field are present
+        for field in dag_detail_fields:
+            assert field in out
+
+        # Check if identifying values are present
+        dag_details_values = ["airflow", "airflow/example_dags/example_complex.py", "16", "example_complex"]
+
+        for value in dag_details_values:
+            assert value in out
+
     @conf_vars({("core", "load_examples"): "true"})
     def test_cli_list_dags(self):
         args = self.parser.parse_args(["dags", "list", "--output", "yaml"])