Add support for JSON output format

Author: chrysle

piptools/scripts/compile.py

@@ -77,6 +77,7 @@ def _determine_linesep(
 @options.color
 @options.verbose
 @options.quiet
+@options.json
 @options.dry_run
 @options.pre
 @options.rebuild
@@ -122,6 +123,7 @@ def cli(
     color: bool | None,
     verbose: int,
     quiet: int,
+    json: bool,
     dry_run: bool,
     pre: bool,
     rebuild: bool,
@@ -506,6 +508,7 @@ def cli(
         cast(BinaryIO, output_file),
         click_ctx=ctx,
         dry_run=dry_run,
+        json_output=json,
         emit_header=header,
         emit_index_url=emit_index_url,
         emit_trusted_host=emit_trusted_host,

piptools/scripts/options.py

@@ -52,6 +52,10 @@ def _get_default_option(option_name: str) -> Any:
     help="Give less output",
 )
 
+json = click.option(
+    "-j", "--json", is_flag=True, default=False, help="Emit JSON output"
+)
+
 dry_run = click.option(
     "-n",
     "--dry-run",

piptools/utils.py

@@ -53,6 +53,7 @@
     "--cache-dir",
     "--no-reuse-hashes",
     "--no-config",
+    "--json",
 }
 
 # Set of option that are only negative, i.e. --no-<option>
@@ -343,7 +344,7 @@ def get_compile_command(click_ctx: click.Context) -> str:
         - removing values that are already default
         - sorting the arguments
         - removing one-off arguments like '--upgrade'
-        - removing arguments that don't change build behaviour like '--verbose'
+        - removing arguments that don't change build behaviour like '--verbose' or '--json'
     """
     from piptools.scripts.compile import cli
 

piptools/writer.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import io
+import json
 import os
 import re
 import sys
@@ -79,6 +80,7 @@ def __init__(
         dst_file: BinaryIO,
         click_ctx: Context,
         dry_run: bool,
+        json_output: bool,
         emit_header: bool,
         emit_index_url: bool,
         emit_trusted_host: bool,
@@ -99,6 +101,7 @@ def __init__(
         self.dst_file = dst_file
         self.click_ctx = click_ctx
         self.dry_run = dry_run
+        self.json_output = json_output
         self.emit_header = emit_header
         self.emit_index_url = emit_index_url
         self.emit_trusted_host = emit_trusted_host
@@ -173,14 +176,62 @@ def write_flags(self) -> Iterator[str]:
         if emitted:
             yield ""
 
-    def _iter_lines(
+    def _get_json(
+        self,
+        ireq: InstallRequirement,
+        line: str,
+        hashes: dict[InstallRequirement, set[str]] | None = None,
+        unsafe: bool = False,
+    ) -> dict[str, str]:
+        """Get a JSON representation for an ``InstallRequirement``."""
+        if hashes:
+            ireq_hashes = hashes.get(ireq)
+            if ireq_hashes:
+                assert isinstance(ireq_hashes, set)
+                output_hashes = list(ireq_hashes)
+            else:
+                output_hashes = []
+        hashable = True
+        if ireq.link:
+            if ireq.link.is_vcs or (ireq.link.is_file and ireq.link.is_existing_dir()):
+                hashable = False
+        markers = ""
+        if ireq.markers:
+            markers = str(ireq.markers)
+        # Retrieve parent requirements from constructed line
+        splitted_line = [m.strip() for m in unstyle(line).split("#")]
+        try:
+            via = splitted_line[splitted_line.index("via") + 1 :]
+        except ValueError:
+            via = [splitted_line[-1][len("via ") :]]
+            if via[0].startswith("-r"):
+                req_files = re.split(r"\s|,", via[0])
+                del req_files[0]
+                via = ["-r"]
+                for req_file in req_files:
+                    via.append(os.path.abspath(req_file))
+        ireq_json = {
+            "name": ireq.name,
+            "version": str(ireq.specifier).lstrip("=="),
+            "requirement": str(ireq.req),
+            "via": via,
+            "line": unstyle(line),
+            "hashable": hashable,
+            "editable": ireq.editable,
+            "hashes": output_hashes,
+            "markers": markers,
+            "unsafe": unsafe,
+        }
+        return ireq_json
+
+    def _iter_ireqs(
         self,
         results: set[InstallRequirement],
         unsafe_requirements: set[InstallRequirement],
         unsafe_packages: set[str],
         markers: dict[str, Marker],
         hashes: dict[InstallRequirement, set[str]] | None = None,
-    ) -> Iterator[str]:
+    ) -> Iterator[str, dict[str, str]]:
         # default values
         unsafe_packages = unsafe_packages if self.allow_unsafe else set()
         hashes = hashes or {}
@@ -191,12 +242,11 @@ def _iter_lines(
         has_hashes = hashes and any(hash for hash in hashes.values())
 
         yielded = False
-
         for line in self.write_header():
-            yield line
+            yield line, {}
             yielded = True
         for line in self.write_flags():
-            yield line
+            yield line, {}
             yielded = True
 
         unsafe_requirements = unsafe_requirements or {
@@ -207,36 +257,36 @@ def _iter_lines(
         if packages:
             for ireq in sorted(packages, key=self._sort_key):
                 if has_hashes and not hashes.get(ireq):
-                    yield MESSAGE_UNHASHED_PACKAGE
+                    yield MESSAGE_UNHASHED_PACKAGE, {}
                     warn_uninstallable = True
                 line = self._format_requirement(
                     ireq, markers.get(key_from_ireq(ireq)), hashes=hashes
                 )
-                yield line
+                yield line, self._get_json(ireq, line, hashes=hashes)
             yielded = True
 
         if unsafe_requirements:
-            yield ""
+            yield "", {}
             yielded = True
             if has_hashes and not self.allow_unsafe:
-                yield MESSAGE_UNSAFE_PACKAGES_UNPINNED
+                yield MESSAGE_UNSAFE_PACKAGES_UNPINNED, {}
                 warn_uninstallable = True
             else:
-                yield MESSAGE_UNSAFE_PACKAGES
+                yield MESSAGE_UNSAFE_PACKAGES, {}
 
             for ireq in sorted(unsafe_requirements, key=self._sort_key):
                 ireq_key = key_from_ireq(ireq)
                 if not self.allow_unsafe:
-                    yield comment(f"# {ireq_key}")
+                    yield comment(f"# {ireq_key}"), {}
                 else:
                     line = self._format_requirement(
                         ireq, marker=markers.get(ireq_key), hashes=hashes
                     )
-                    yield line
+                    yield line, self._get_json(ireq, line, unsafe=True)
 
         # Yield even when there's no real content, so that blank files are written
         if not yielded:
-            yield ""
+            yield "", {}
 
         if warn_uninstallable:
             log.warning(MESSAGE_UNINSTALLABLE)
@@ -249,27 +299,33 @@ def write(
         markers: dict[str, Marker],
         hashes: dict[InstallRequirement, set[str]] | None,
     ) -> None:
-        if not self.dry_run:
+        output_structure = []
+        if not self.dry_run or self.json_output:
             dst_file = io.TextIOWrapper(
                 self.dst_file,
                 encoding="utf8",
                 newline=self.linesep,
                 line_buffering=True,
             )
         try:
-            for line in self._iter_lines(
+            for line, ireq in self._iter_ireqs(
                 results, unsafe_requirements, unsafe_packages, markers, hashes
             ):
                 if self.dry_run:
                     # Bypass the log level to always print this during a dry run
                     log.log(line)
                 else:
-                    log.info(line)
+                    if not self.json_output:
+                        log.info(line)
                     dst_file.write(unstyle(line))
                     dst_file.write("\n")
+                if self.json_output and ireq:
+                    output_structure.append(ireq)
         finally:
-            if not self.dry_run:
+            if not self.dry_run or self.json_output:
                 dst_file.detach()
+            if self.json_output:
+                print(json.dumps(output_structure, indent=4))
 
     def _format_requirement(
         self,

tests/test_writer.py

@@ -34,6 +34,7 @@ def writer(tmpdir_cwd):
             dst_file=ctx.params["output_file"],
             click_ctx=ctx,
             dry_run=True,
+            json_output=False,
             emit_header=True,
             emit_index_url=True,
             emit_trusted_host=True,
@@ -108,11 +109,11 @@ def test_format_requirement_environment_marker(from_line, writer):
 
 
 @pytest.mark.parametrize("allow_unsafe", ((True,), (False,)))
-def test_iter_lines__unsafe_dependencies(writer, from_line, allow_unsafe):
+def test_iter_ireqs__unsafe_dependencies(writer, from_line, allow_unsafe):
     writer.allow_unsafe = allow_unsafe
     writer.emit_header = False
 
-    lines = writer._iter_lines(
+    lines = writer._iter_ireqs(
         {from_line("test==1.2")},
         {from_line("setuptools==1.10.0")},
         unsafe_packages=set(),
@@ -128,14 +129,14 @@ def test_iter_lines__unsafe_dependencies(writer, from_line, allow_unsafe):
     assert tuple(lines) == expected_lines
 
 
-def test_iter_lines__unsafe_with_hashes(capsys, writer, from_line):
+def test_iter_ireqs__unsafe_with_hashes(capsys, writer, from_line):
     writer.allow_unsafe = False
     writer.emit_header = False
     ireqs = [from_line("test==1.2")]
     unsafe_ireqs = [from_line("setuptools==1.10.0")]
     hashes = {ireqs[0]: {"FAKEHASH"}, unsafe_ireqs[0]: set()}
 
-    lines = writer._iter_lines(
+    lines = writer._iter_ireqs(
         ireqs, unsafe_ireqs, unsafe_packages=set(), markers={}, hashes=hashes
     )
 
@@ -151,13 +152,13 @@ def test_iter_lines__unsafe_with_hashes(capsys, writer, from_line):
     assert captured.err.strip() == MESSAGE_UNINSTALLABLE
 
 
-def test_iter_lines__hash_missing(capsys, writer, from_line):
+def test_iter_ireqs__hash_missing(capsys, writer, from_line):
     writer.allow_unsafe = False
     writer.emit_header = False
     ireqs = [from_line("test==1.2"), from_line("file:///example/#egg=example")]
     hashes = {ireqs[0]: {"FAKEHASH"}, ireqs[1]: set()}
 
-    lines = writer._iter_lines(
+    lines = writer._iter_ireqs(
         ireqs,
         hashes=hashes,
         unsafe_requirements=set(),
@@ -176,7 +177,7 @@ def test_iter_lines__hash_missing(capsys, writer, from_line):
     assert captured.err.strip() == MESSAGE_UNINSTALLABLE
 
 
-def test_iter_lines__no_warn_if_only_unhashable_packages(writer, from_line):
+def test_iter_ireqs__no_warn_if_only_unhashable_packages(writer, from_line):
     """
     There shouldn't be MESSAGE_UNHASHED_PACKAGE warning if there are only unhashable
     packages. See GH-1101.
@@ -189,7 +190,7 @@ def test_iter_lines__no_warn_if_only_unhashable_packages(writer, from_line):
     ]
     hashes = {ireq: set() for ireq in ireqs}
 
-    lines = writer._iter_lines(
+    lines = writer._iter_ireqs(
         ireqs,
         hashes=hashes,
         unsafe_requirements=set(),
@@ -418,7 +419,7 @@ def test_write_order(writer, from_line):
         "package-b==2.3.4",
         "package2==7.8.9",
     ]
-    result = writer._iter_lines(
+    result = writer._iter_ireqs(
         packages, unsafe_requirements=set(), unsafe_packages=set(), markers={}
     )
     assert list(result) == expected_lines