feat: dvc stage list --json

dvc/commands/stage.py

@@ -65,6 +65,43 @@ def prepare_stages_data(
     }
 
 
+def _serialize_stage_for_json(stage: "Stage") -> dict:
+    from typing import Any
+
+    from dvc.stage.utils import split_params_deps
+
+    param_deps, other_deps = split_params_deps(stage)
+    deps = [dep.def_path for dep in other_deps]
+    params: dict[str, Any] = {}
+    for param_dep in param_deps:
+        param_values: Any = param_dep.hash_info.value if param_dep.hash_info else {}
+        if isinstance(param_values, dict) and param_values:
+            params[param_dep.def_path] = param_values
+
+    outs, metrics, plots = [], [], []
+    for out in stage.outs:
+        if out.metric:
+            metrics.append(out.def_path)
+        elif out.plot:
+            plots.append(out.def_path)
+        else:
+            outs.append(out.def_path)
+
+    return {
+        "cmd": stage.cmd,
+        "deps": deps,
+        "outs": outs,
+        "metrics": metrics,
+        "plots": plots,
+        "params": params,
+        "desc": stage.desc,
+    }
+
+
+def prepare_stages_data_json(stages: Iterable["Stage"]) -> dict[str, dict]:
+    return {stage.addressing: _serialize_stage_for_json(stage) for stage in stages}
+
+
 class CmdStageList(CmdBase):
     def _get_stages(self) -> Iterable["Stage"]:
         if self.args.all:
@@ -91,8 +128,17 @@ def log_error(relpath: str, exc: Exception):
         self.repo.stage_collection_error_handler = log_error
 
         stages = self._get_stages()
-        data = prepare_stages_data(stages, description=not self.args.name_only)
-        ui.table(list(data.items()))
+
+        if self.args.json:
+            ui.write_json(prepare_stages_data_json(stages))
+        else:
+            ui.table(
+                list(
+                    prepare_stages_data(
+                        stages, description=not self.args.name_only
+                    ).items()
+                )
+            )
 
         return 0
 
@@ -351,4 +397,10 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="List only stage names.",
     )
+    stage_list_parser.add_argument(
+        "--json",
+        action="store_true",
+        default=False,
+        help="Show output in JSON format.",
+    )
     stage_list_parser.set_defaults(func=CmdStageList)

tests/func/test_stage_list_json.py

@@ -0,0 +1,143 @@
+import json
+
+import pytest
+
+from dvc.cli import main
+
+
+@pytest.fixture
+def simple_stage(tmp_dir, dvc):
+    tmp_dir.gen("train.py", "print('training')")
+    tmp_dir.gen("data.csv", "a,b,c")
+    (tmp_dir / "dvc.yaml").dump(
+        {
+            "stages": {
+                "train": {
+                    "cmd": "python train.py",
+                    "deps": ["data.csv"],
+                    "outs": ["model.pkl"],
+                    "metrics": [{"metrics.json": {"cache": False}}],
+                    "desc": "Train the model",
+                }
+            }
+        }
+    )
+    return dvc
+
+
+def test_stage_list_json_simple(simple_stage, tmp_dir, capsys):
+    assert main(["stage", "list", "--json"]) == 0
+
+    out, _ = capsys.readouterr()
+    result = json.loads(out)
+
+    assert "train" in result
+    assert result["train"]["cmd"] == "python train.py"
+    assert "data.csv" in result["train"]["deps"]
+    assert "model.pkl" in result["train"]["outs"]
+    assert "metrics.json" in result["train"]["metrics"]
+    assert result["train"]["desc"] == "Train the model"
+
+
+@pytest.fixture
+def interpolated_stage(tmp_dir, dvc):
+    tmp_dir.gen("train.py", "print('training')")
+    (tmp_dir / "params.yaml").dump({"train": {"lr": 0.001, "epochs": 100}})
+    (tmp_dir / "dvc.yaml").dump(
+        {
+            "stages": {
+                "train": {
+                    "cmd": "python train.py --lr ${train.lr} --epochs ${train.epochs}",
+                    "params": ["train.lr", "train.epochs"],
+                }
+            }
+        }
+    )
+    return dvc
+
+
+def test_stage_list_json_interpolated_params(interpolated_stage, tmp_dir, capsys):
+    assert main(["stage", "list", "--json"]) == 0
+
+    out, _ = capsys.readouterr()
+    result = json.loads(out)
+
+    assert "train" in result
+    assert result["train"]["cmd"] == "python train.py --lr 0.001 --epochs 100"
+
+
+@pytest.fixture
+def matrix_stage(tmp_dir, dvc):
+    tmp_dir.gen("train.py", "print('training')")
+    (tmp_dir / "dvc.yaml").dump(
+        {
+            "stages": {
+                "train": {
+                    "matrix": {"lr": [0.001, 0.01], "epochs": [10, 100]},
+                    "cmd": "python train.py --lr ${item.lr} --epochs ${item.epochs}",
+                }
+            }
+        }
+    )
+    return dvc
+
+
+def test_stage_list_json_matrix_stage(matrix_stage, tmp_dir, capsys):
+    assert main(["stage", "list", "--json"]) == 0
+
+    out, _ = capsys.readouterr()
+    result = json.loads(out)
+
+    assert len(result) == 4
+    for stage_name, stage_data in result.items():
+        assert stage_name.startswith("train@")
+        assert "python train.py" in stage_data["cmd"]
+        assert "--lr" in stage_data["cmd"]
+        assert "--epochs" in stage_data["cmd"]
+
+
+@pytest.fixture
+def foreach_stage(tmp_dir, dvc):
+    tmp_dir.gen("process.py", "print('processing')")
+    (tmp_dir / "dvc.yaml").dump(
+        {
+            "stages": {
+                "process": {
+                    "foreach": ["a", "b", "c"],
+                    "do": {"cmd": "python process.py --file ${item}"},
+                }
+            }
+        }
+    )
+    return dvc
+
+
+def test_stage_list_json_foreach_stage(foreach_stage, tmp_dir, capsys):
+    assert main(["stage", "list", "--json"]) == 0
+
+    out, _ = capsys.readouterr()
+    result = json.loads(out)
+
+    expected_stages = ["process@a", "process@b", "process@c"]
+    for stage_name in expected_stages:
+        assert stage_name in result
+        assert "python process.py" in result[stage_name]["cmd"]
+
+
+def test_stage_list_json_with_target(simple_stage, tmp_dir, capsys):
+    assert main(["stage", "list", "--json", "train"]) == 0
+
+    out, _ = capsys.readouterr()
+    result = json.loads(out)
+
+    assert "train" in result
+    assert len(result) == 1
+
+
+def test_stage_list_json_all_flag(simple_stage, tmp_dir, capsys):
+    assert main(["stage", "list", "--json", "--all"]) == 0
+
+    out, _ = capsys.readouterr()
+    result = json.loads(out)
+
+    assert "train" in result