Add support for JSON output format

piptools/scripts/compile.py

@@ -28,6 +28,7 @@
     install_req_from_line,
     is_pinned_requirement,
     key_from_ireq,
+    render_requirements_json_txt,
 )
 from ..writer import OutputWriter
 from . import options
@@ -41,6 +42,7 @@
 )
 DEFAULT_REQUIREMENTS_FILE = "requirements.in"
 DEFAULT_REQUIREMENTS_OUTPUT_FILE = "requirements.txt"
+DEFAULT_REQUIREMENTS_OUTPUT_FILE_JSON = "requirements.json"
 METADATA_FILENAMES = frozenset({"setup.py", "setup.cfg", "pyproject.toml"})
 
 
@@ -82,6 +84,7 @@ def _determine_linesep(
 @options.color
 @options.verbose
 @options.quiet
+@options.json
 @options.dry_run
 @options.pre
 @options.rebuild
@@ -127,6 +130,7 @@ def cli(
     color: bool | None,
     verbose: int,
     quiet: int,
+    json: bool,
     dry_run: bool,
     pre: bool,
     rebuild: bool,
@@ -218,10 +222,16 @@ def cli(
         # An output file must be provided for stdin
         if src_files == ("-",):
             raise click.BadParameter("--output-file is required if input is from stdin")
-        # Use default requirements output file if there is a setup.py the source file
+        # Use default requirements output file if the source file is a recognized
+        # packaging metadata file
         elif os.path.basename(src_files[0]) in METADATA_FILENAMES:
             file_name = os.path.join(
-                os.path.dirname(src_files[0]), DEFAULT_REQUIREMENTS_OUTPUT_FILE
+                os.path.dirname(src_files[0]),
+                (
+                    DEFAULT_REQUIREMENTS_OUTPUT_FILE_JSON
+                    if json
+                    else DEFAULT_REQUIREMENTS_OUTPUT_FILE
+                ),
             )
         # An output file must be provided if there are multiple source files
         elif len(src_files) > 1:
@@ -311,11 +321,16 @@ def cli(
                 "as any existing content is truncated."
             )
 
+        if json:
+            # Render contents of JSON output file to a temporary requirements
+            # file in text format in order to make it readable by ``pip``
+            tmpfile_name = render_requirements_json_txt(output_file.name)
+
         # Use a temporary repository to ensure outdated(removed) options from
         # existing requirements.txt wouldn't get into the current repository.
         tmp_repository = PyPIRepository(pip_args, cache_dir=cache_dir)
         ireqs = parse_requirements(
-            output_file.name,
+            tmpfile_name if json else output_file.name,
             finder=tmp_repository.finder,
             session=tmp_repository.session,
             options=tmp_repository.options,
@@ -514,6 +529,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

@@ -9,6 +9,7 @@
 import re
 import shlex
 import sys
+import tempfile
 from pathlib import Path
 from typing import Any, Callable, Iterable, Iterator, TypeVar, cast
 
@@ -58,6 +59,7 @@
     "--cache-dir",
     "--no-reuse-hashes",
     "--no-config",
+    "--json",
 }
 
 # Set of option that are only negative, i.e. --no-<option>
@@ -352,7 +354,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
 
@@ -771,3 +773,18 @@ def is_path_relative_to(path1: Path, path2: Path) -> bool:
     except ValueError:
         return False
     return True
+
+
+def render_requirements_json_txt(filename: str) -> str:
+    """Render a given ``requirements.json`` file to a temporary
+    ``requirements.txt`` file and return its name.
+    """
+    with open(filename, encoding="utf-8") as f:
+        reqs = json.load(f)
+    tmpfile = tempfile.NamedTemporaryFile(mode="w+t", encoding="utf-8", delete=False)
+    for req in reqs:
+        tmpfile.write(req["line"])
+        tmpfile.write("\n")
+    tmpfile.flush()
+
+    return tmpfile.name

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
@@ -191,14 +194,14 @@ def write_flags(self) -> Iterator[str]:
         if emitted:
             yield ""
 
-    def _iter_lines(
+    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] | Iterator[dict[str, str | list[str]]]:
         # default values
         unsafe_packages = unsafe_packages if self.allow_unsafe else set()
         hashes = hashes or {}
@@ -209,13 +212,13 @@ def _iter_lines(
         has_hashes = hashes and any(hash for hash in hashes.values())
 
         yielded = False
-
-        for line in self.write_header():
-            yield line
-            yielded = True
-        for line in self.write_flags():
-            yield line
-            yielded = True
+        if not self.json_output:
+            for line in self.write_header():
+                yield line
+                yielded = True
+            for line in self.write_flags():
+                yield line
+                yielded = True
 
         unsafe_requirements = unsafe_requirements or {
             r for r in results if r.name in unsafe_packages
@@ -224,33 +227,35 @@ def _iter_lines(
 
         if packages:
             for ireq in sorted(packages, key=self._sort_key):
-                if has_hashes and not hashes.get(ireq):
+                if has_hashes and not hashes.get(ireq) and not self.json_output:
                     yield MESSAGE_UNHASHED_PACKAGE
                     warn_uninstallable = True
-                line = self._format_requirement(
+                formatted_req = self._format_requirement(
                     ireq, markers.get(key_from_ireq(ireq)), hashes=hashes
                 )
-                yield line
+                yield formatted_req
             yielded = True
 
         if unsafe_requirements:
-            yield ""
+
+            if not self.json_output:
+                yield ""
             yielded = True
-            if has_hashes and not self.allow_unsafe:
+            if has_hashes and not self.allow_unsafe and not self.json_output:
                 yield MESSAGE_UNSAFE_PACKAGES_UNPINNED
                 warn_uninstallable = True
-            else:
+            elif not self.json_output:
                 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:
+                if not self.allow_unsafe and not self.json_output:
                     yield comment(f"# {ireq_key}")
                 else:
-                    line = self._format_requirement(
+                    formatted_req = self._format_requirement(
                         ireq, marker=markers.get(ireq_key), hashes=hashes
                     )
-                    yield line
+                    yield formatted_req
 
         # Yield even when there's no real content, so that blank files are written
         if not yielded:
@@ -267,6 +272,7 @@ def write(
         markers: dict[str, Marker],
         hashes: dict[InstallRequirement, set[str]] | None,
     ) -> None:
+        output_structure = []
         if not self.dry_run:
             dst_file = io.TextIOWrapper(
                 self.dst_file,
@@ -275,17 +281,25 @@ def write(
                 line_buffering=True,
             )
         try:
-            for line in self._iter_lines(
+            for formatted_req in self._iter_ireqs(
                 results, unsafe_requirements, unsafe_packages, markers, hashes
             ):
-                if self.dry_run:
+                if self.dry_run and not self.json_output:
                     # Bypass the log level to always print this during a dry run
-                    log.log(line)
+                    assert isinstance(formatted_req, str)
+                    log.log(formatted_req)
                 else:
-                    log.info(line)
-                    dst_file.write(unstyle(line))
-                    dst_file.write("\n")
+                    if not self.json_output:
+                        assert isinstance(formatted_req, str)
+                        log.info(formatted_req)
+                        dst_file.write(unstyle(formatted_req))
+                        dst_file.write("\n")
+                    else:
+                        output_structure.append(formatted_req)
         finally:
+            if self.json_output:
+                json.dump(output_structure, dst_file, indent=4)
+                print(json.dumps(output_structure, indent=4))
             if not self.dry_run:
                 dst_file.detach()
 
@@ -294,16 +308,18 @@ def _format_requirement(
         ireq: InstallRequirement,
         marker: Marker | None = None,
         hashes: dict[InstallRequirement, set[str]] | None = None,
-    ) -> str:
+        unsafe: bool = False,
+    ) -> str | dict[str, str | list[str]]:
+        """Format a given ``InstallRequirement``.
+
+        :returns: A line or a JSON structure to be written to the output file.
+        """
         ireq_hashes = (hashes if hashes is not None else {}).get(ireq)
 
         line = format_requirement(ireq, marker=marker, hashes=ireq_hashes)
         if self.strip_extras:
             line = strip_extras(line)
 
-        if not self.annotate:
-            return line
-
         # Annotate what packages or reqs-ins this package is required by
         required_by = set()
         if hasattr(ireq, "_source_ireqs"):
@@ -335,6 +351,43 @@ def _format_requirement(
                 annotation = strip_extras(annotation)
             # 24 is one reasonable column size to use here, that we've used in the past
             lines = f"{line:24}{sep}{comment(annotation)}".splitlines()
-            line = "\n".join(ln.rstrip() for ln in lines)
+            if self.annotate:
+                line = "\n".join(ln.rstrip() for ln in lines)
+
+        if self.json_output:
+            hashable = True
+            if ireq.link:
+                if ireq.link.is_vcs or (
+                    ireq.link.is_file and ireq.link.is_existing_dir()
+                ):
+                    hashable = False
+            output_marker = ""
+            if marker:
+                output_marker = str(marker)
+            via = []
+            for parent_req in required_by:
+                if parent_req.startswith("-r "):
+                    # Ensure paths to requirements files given are absolute
+                    reqs_in_path = os.path.abspath(parent_req[len("-r ") :])
+                    via.append(f"-r {reqs_in_path}")
+                else:
+                    via.append(parent_req)
+            output_hashes = []
+            if ireq_hashes:
+                output_hashes = list(ireq_hashes)
+
+            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,
+                "marker": output_marker,
+                "unsafe": unsafe,
+            }
+            return ireq_json
 
         return line