Adding upstream version 124.0.1.upstream/124.0.1

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 243 insertions, 0 deletions

diff --git a/third_party/python/pip_tools/piptools/writer.py b/third_party/python/pip_tools/piptools/writer.py
new file mode 100644
index 0000000000..515df198eb
--- /dev/null
+++ b/third_party/python/pip_tools/piptools/writer.py
@@ -0,0 +1,243 @@
+from __future__ import unicode_literals
+
+import os
+import re
+from itertools import chain
+
+from pip._vendor import six
+
+from .click import unstyle
+from .logging import log
+from .utils import (
+    UNSAFE_PACKAGES,
+    comment,
+    dedup,
+    format_requirement,
+    get_compile_command,
+    key_from_ireq,
+)
+
+MESSAGE_UNHASHED_PACKAGE = comment(
+    "# WARNING: pip install will require the following package to be hashed."
+    "\n# Consider using a hashable URL like "
+    "https://github.com/jazzband/pip-tools/archive/SOMECOMMIT.zip"
+)
+
+MESSAGE_UNSAFE_PACKAGES_UNPINNED = comment(
+    "# WARNING: The following packages were not pinned, but pip requires them to be"
+    "\n# pinned when the requirements file includes hashes. "
+    "Consider using the --allow-unsafe flag."
+)
+
+MESSAGE_UNSAFE_PACKAGES = comment(
+    "# The following packages are considered to be unsafe in a requirements file:"
+)
+
+MESSAGE_UNINSTALLABLE = (
+    "The generated requirements file may be rejected by pip install. "
+    "See # WARNING lines for details."
+)
+
+
+strip_comes_from_line_re = re.compile(r" \(line \d+\)$")
+
+
+def _comes_from_as_string(ireq):
+    if isinstance(ireq.comes_from, six.string_types):
+        return strip_comes_from_line_re.sub("", ireq.comes_from)
+    return key_from_ireq(ireq.comes_from)
+
+
+class OutputWriter(object):
+    def __init__(
+        self,
+        src_files,
+        dst_file,
+        click_ctx,
+        dry_run,
+        emit_header,
+        emit_index_url,
+        emit_trusted_host,
+        annotate,
+        generate_hashes,
+        default_index_url,
+        index_urls,
+        trusted_hosts,
+        format_control,
+        allow_unsafe,
+        find_links,
+        emit_find_links,
+    ):
+        self.src_files = src_files
+        self.dst_file = dst_file
+        self.click_ctx = click_ctx
+        self.dry_run = dry_run
+        self.emit_header = emit_header
+        self.emit_index_url = emit_index_url
+        self.emit_trusted_host = emit_trusted_host
+        self.annotate = annotate
+        self.generate_hashes = generate_hashes
+        self.default_index_url = default_index_url
+        self.index_urls = index_urls
+        self.trusted_hosts = trusted_hosts
+        self.format_control = format_control
+        self.allow_unsafe = allow_unsafe
+        self.find_links = find_links
+        self.emit_find_links = emit_find_links
+
+    def _sort_key(self, ireq):
+        return (not ireq.editable, str(ireq.req).lower())
+
+    def write_header(self):
+        if self.emit_header:
+            yield comment("#")
+            yield comment("# This file is autogenerated by pip-compile")
+            yield comment("# To update, run:")
+            yield comment("#")
+            compile_command = os.environ.get(
+                "CUSTOM_COMPILE_COMMAND"
+            ) or get_compile_command(self.click_ctx)
+            yield comment("#    {}".format(compile_command))
+            yield comment("#")
+
+    def write_index_options(self):
+        if self.emit_index_url:
+            for index, index_url in enumerate(dedup(self.index_urls)):
+                if index_url.rstrip("/") == self.default_index_url:
+                    continue
+                flag = "--index-url" if index == 0 else "--extra-index-url"
+                yield "{} {}".format(flag, index_url)
+
+    def write_trusted_hosts(self):
+        if self.emit_trusted_host:
+            for trusted_host in dedup(self.trusted_hosts):
+                yield "--trusted-host {}".format(trusted_host)
+
+    def write_format_controls(self):
+        for nb in dedup(sorted(self.format_control.no_binary)):
+            yield "--no-binary {}".format(nb)
+        for ob in dedup(sorted(self.format_control.only_binary)):
+            yield "--only-binary {}".format(ob)
+
+    def write_find_links(self):
+        if self.emit_find_links:
+            for find_link in dedup(self.find_links):
+                yield "--find-links {}".format(find_link)
+
+    def write_flags(self):
+        emitted = False
+        for line in chain(
+            self.write_index_options(),
+            self.write_find_links(),
+            self.write_trusted_hosts(),
+            self.write_format_controls(),
+        ):
+            emitted = True
+            yield line
+        if emitted:
+            yield ""
+
+    def _iter_lines(self, results, unsafe_requirements=None, markers=None, hashes=None):
+        # default values
+        unsafe_requirements = unsafe_requirements or []
+        markers = markers or {}
+        hashes = hashes or {}
+
+        # Check for unhashed or unpinned packages if at least one package does have
+        # hashes, which will trigger pip install's --require-hashes mode.
+        warn_uninstallable = False
+        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
+
+        unsafe_requirements = (
+            {r for r in results if r.name in UNSAFE_PACKAGES}
+            if not unsafe_requirements
+            else unsafe_requirements
+        )
+        packages = {r for r in results if r.name not in UNSAFE_PACKAGES}
+
+        if packages:
+            packages = sorted(packages, key=self._sort_key)
+            for ireq in packages:
+                if has_hashes and not hashes.get(ireq):
+                    yield MESSAGE_UNHASHED_PACKAGE
+                    warn_uninstallable = True
+                line = self._format_requirement(
+                    ireq, markers.get(key_from_ireq(ireq)), hashes=hashes
+                )
+                yield line
+            yielded = True
+
+        if unsafe_requirements:
+            unsafe_requirements = sorted(unsafe_requirements, key=self._sort_key)
+            yield ""
+            yielded = True
+            if has_hashes and not self.allow_unsafe:
+                yield MESSAGE_UNSAFE_PACKAGES_UNPINNED
+                warn_uninstallable = True
+            else:
+                yield MESSAGE_UNSAFE_PACKAGES
+
+            for ireq in unsafe_requirements:
+                ireq_key = key_from_ireq(ireq)
+                if not self.allow_unsafe:
+                    yield comment("# {}".format(ireq_key))
+                else:
+                    line = self._format_requirement(
+                        ireq, marker=markers.get(ireq_key), hashes=hashes
+                    )
+                    yield line
+
+        # Yield even when there's no real content, so that blank files are written
+        if not yielded:
+            yield ""
+
+        if warn_uninstallable:
+            log.warning(MESSAGE_UNINSTALLABLE)
+
+    def write(self, results, unsafe_requirements, markers, hashes):
+
+        for line in self._iter_lines(results, unsafe_requirements, markers, hashes):
+            log.info(line)
+            if not self.dry_run:
+                self.dst_file.write(unstyle(line).encode("utf-8"))
+                self.dst_file.write(os.linesep.encode("utf-8"))
+
+    def _format_requirement(self, ireq, marker=None, hashes=None):
+        ireq_hashes = (hashes if hashes is not None else {}).get(ireq)
+
+        line = format_requirement(ireq, marker=marker, hashes=ireq_hashes)
+
+        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"):
+            required_by |= {
+                _comes_from_as_string(src_ireq)
+                for src_ireq in ireq._source_ireqs
+                if src_ireq.comes_from
+            }
+        elif ireq.comes_from:
+            required_by.add(_comes_from_as_string(ireq))
+        if required_by:
+            required_by = sorted(required_by)
+            if len(required_by) == 1:
+                source = required_by[0]
+                annotation = "    # via " + source
+            else:
+                annotation_lines = ["    # via"]
+                for source in required_by:
+                    annotation_lines.append("    #   " + source)
+                annotation = "\n".join(annotation_lines)
+            line = "{}\n{}".format(line, comment(annotation))
+        return line