initial wip to get pipeline json from a specific commit

.isort.cfg

@@ -1,2 +1,2 @@
 [settings]
-known_third_party = more_itertools,pytest,setuptools
+known_third_party = click,kedro,more_itertools,pytest,rich,setuptools

kedro_diff/cli.py

@@ -0,0 +1,112 @@
+"""kedro_diff cli module."""
+import logging
+from typing import IO, TYPE_CHECKING, Tuple
+
+import click
+from kedro.framework.session import KedroSession
+from rich import print
+
+if TYPE_CHECKING:
+    from kedro.framework.startup import ProjectMetadata
+
+__version__ = "0.0.0"
+
+
+def silent_loggers() -> None:
+    """All logs need to be silent in order for a clean kedro diff output."""
+    known_kedro_loggers = [
+        "ProfileTimeTransformer",
+        "hooks_handler",
+        "kedro.__init__",
+        "kedro",
+        "kedro.config",
+        "kedro.config.config",
+        "kedro.extras.decorators.memory_profiler",
+        "kedro.framework.cli",
+        "kedro.framework.session.session",
+        "kedro.framework.session.store",
+        "kedro.framework.session",
+        "kedro.io.cached_dataset",
+        "kedro.io.data_catalog",
+        "kedro.io",
+        "kedro.journal",
+        "kedro.pipeline",
+        "kedro.pipeline.decorators",
+        "kedro.pipeline.node",
+        "kedro.pipeline.pipeline",
+        "kedro.runner",
+        "kedro.runner.runner",
+        "kedro.versioning.journal",
+        "py4",
+    ]
+    for logger in [
+        *known_kedro_loggers,
+        *list(logging.root.manager.loggerDict.keys()),  # type: ignore
+    ]:
+        logging.getLogger(logger).setLevel(logging.ERROR)
+
+
+@click.group(name="kedro-diff")
+@click.version_option(__version__, "-V", "--version", help="Prints version and exits")
+def cli() -> None:
+    """Kedro diff cli group."""
+    pass
+
+
+@cli.command()
+@click.option("-o", "--output", type=click.File("wb"))
+@click.option(
+    "-v",
+    "--verbose",
+    count=True,
+    help="verbosity level, -v enables diff related logs, -vv enables all logs",
+)
+@click.option("-q", "--quiet", is_flag=True, help="runs completely quiet")
+# @click.argument("commit", nargs=-1)
+@click.pass_obj
+def get_json(
+    metadata: "ProjectMetadata", output: IO, verbose: int, quiet: bool
+) -> None:
+    """Get pipeline json from project context."""
+    if quiet:
+        verbose = -1
+    logger = logging.getLogger(__name__)
+    logger.setLevel(logging.INFO)
+    if verbose < 1:
+        logger.setLevel(logging.ERROR)
+
+        # commits = parse_commit(commit)
+
+    if verbose < 2:
+        silent_loggers()
+
+    session = KedroSession.create(metadata.package_name)
+    context = session.load_context()
+
+    pipeline: str = context.pipeline.to_json()  # type: ignore
+    if verbose >= 0:
+        print(pipeline)
+    output.write(pipeline.encode("utf-8"))
+
+
+@cli.command()
+@click.option(
+    "-v",
+    "--verbose",
+    count=True,
+    help="verbosity level, -v enables diff related logs, -vv enables all logs",
+)
+@click.argument("commit", nargs=-1)
+@click.pass_obj
+def diff(metadata: "ProjectMetadata", verbose: int, commit: Tuple[str, ...]) -> None:
+    """Diff two commits."""
+    from kedro_diff.commit_parser import parse_commit
+    from kedro_diff.get_pipelines import to_json
+
+    commit1, commit2 = parse_commit(commit)
+    to_json(metadata.project_path, commit1)
+    to_json(metadata.project_path, commit2)
+
+
+if __name__ == "__main__":
+    cli()

kedro_diff/get_pipelines.py

@@ -0,0 +1,50 @@
+"""get_pipelines.
+
+Get json from a specific commit
+"""
+import logging
+import os
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Callable, Optional, Union
+
+
+def copytree(
+    src: Union[str, Path],
+    dst: Union[str, Path],
+    symlinks: bool = False,
+    ignore: Optional[Callable] = None,
+) -> None:
+    """Copy src director into dst directory."""
+    for item in os.listdir(src):
+        s = os.path.join(src, item)
+        d = os.path.join(dst, item)
+        if os.path.isdir(s):
+            shutil.copytree(s, d, symlinks, ignore)
+        else:
+            shutil.copy2(s, d)
+
+
+def to_json(project_path: Union[str, Path], commit: str, verbose: int = 0) -> None:
+    """Get json from specific commit."""
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        logger = logging.getLogger(__name__)
+        logger.setLevel(logging.INFO)
+        if verbose < 1:
+            logger.setLevel(logging.ERROR)
+        logger.info(f"copying {project_path} into {tmpdirname}")
+        copytree(project_path, tmpdirname)
+        path = (Path() / ".kedro-diff" / (commit + ".json")).absolute()
+        path.parent.mkdir(exist_ok=True)
+        subprocess.call(f"git checkout {commit} --quiet", shell=True, cwd=tmpdirname)
+        subprocess.call(
+            f"kedro get-json --output {path} --quiet", shell=True, cwd=tmpdirname
+        )
+
+
+if __name__ == "__main__":
+    import sys
+
+    to_json(sys.argv[1], sys.argv[2])

setup.cfg

@@ -7,6 +7,9 @@ testpaths =
     tests
 
 [coverage:report]
+omit =
+  kedro_diff/cli.py
+  kedro_diff/get_pipelines.py
 exclude_lines =
     if __name__ == .__main__.:
 [mypy]