We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Incorrect output on pip-compile upgrade with --dry-run and --quiet flags
##### Environment Versions

1. OS Type: any
1. Python version: any
1. pip version:  any
1. pip-tools version: 3.8.0

##### Steps to replicate

1. `echo "six" > requirements.in`
1. `pip-compile`
1. `pip-compile --upgrade --dry-run --quiet`

##### Expected result

- `--dry-run` should print changes to stdout/stderr
- `--quiet` should suppress _"Dry-run, so nothing updated."_ message 

```bash
#
# This file is autogenerated by pip-compile
# To update, run:
#
#    pip-compile
#
six==1.12.0
```

##### Actual result

But currently it behaves the opposite way:

```
Dry-run, so nothing updated.
```

---

Refs: 

- #843 
- https://github.com/jazzband/pip-tools/issues/841#issuecomment-508454216

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `piptools/writer.py`
Content:
```
1 import os
2 import re
3 import sys
4 from itertools import chain
5 from typing import BinaryIO, Dict, Iterable, Iterator, List, Optional, Set, Tuple
6 
7 from click import unstyle
8 from click.core import Context
9 from pip._internal.models.format_control import FormatControl
10 from pip._internal.req.req_install import InstallRequirement
11 from pip._vendor.packaging.markers import Marker
12 
13 from .logging import log
14 from .utils import (
15     UNSAFE_PACKAGES,
16     comment,
17     dedup,
18     format_requirement,
19     get_compile_command,
20     key_from_ireq,
21 )
22 
23 MESSAGE_UNHASHED_PACKAGE = comment(
24     "# WARNING: pip install will require the following package to be hashed."
25     "\n# Consider using a hashable URL like "
26     "https://github.com/jazzband/pip-tools/archive/SOMECOMMIT.zip"
27 )
28 
29 MESSAGE_UNSAFE_PACKAGES_UNPINNED = comment(
30     "# WARNING: The following packages were not pinned, but pip requires them to be"
31     "\n# pinned when the requirements file includes hashes. "
32     "Consider using the --allow-unsafe flag."
33 )
34 
35 MESSAGE_UNSAFE_PACKAGES = comment(
36     "# The following packages are considered to be unsafe in a requirements file:"
37 )
38 
39 MESSAGE_UNINSTALLABLE = (
40     "The generated requirements file may be rejected by pip install. "
41     "See # WARNING lines for details."
42 )
43 
44 
45 strip_comes_from_line_re = re.compile(r" \(line \d+\)$")
46 
47 
48 def _comes_from_as_string(ireq: InstallRequirement) -> str:
49     if isinstance(ireq.comes_from, str):
50         return strip_comes_from_line_re.sub("", ireq.comes_from)
51     return key_from_ireq(ireq.comes_from)
52 
53 
54 def annotation_style_split(required_by: Set[str]) -> str:
55     sorted_required_by = sorted(required_by)
56     if len(sorted_required_by) == 1:
57         source = sorted_required_by[0]
58         annotation = "# via " + source
59     else:
60         annotation_lines = ["# via"]
61         for source in sorted_required_by:
62             annotation_lines.append("    #   " + source)
63         annotation = "\n".join(annotation_lines)
64     return annotation
65 
66 
67 def annotation_style_line(required_by: Set[str]) -> str:
68     return f"# via {', '.join(sorted(required_by))}"
69 
70 
71 class OutputWriter:
72     def __init__(
73         self,
74         dst_file: BinaryIO,
75         click_ctx: Context,
76         dry_run: bool,
77         emit_header: bool,
78         emit_index_url: bool,
79         emit_trusted_host: bool,
80         annotate: bool,
81         annotation_style: str,
82         strip_extras: bool,
83         generate_hashes: bool,
84         default_index_url: str,
85         index_urls: Iterable[str],
86         trusted_hosts: Iterable[str],
87         format_control: FormatControl,
88         allow_unsafe: bool,
89         find_links: List[str],
90         emit_find_links: bool,
91         emit_options: bool,
92     ) -> None:
93         self.dst_file = dst_file
94         self.click_ctx = click_ctx
95         self.dry_run = dry_run
96         self.emit_header = emit_header
97         self.emit_index_url = emit_index_url
98         self.emit_trusted_host = emit_trusted_host
99         self.annotate = annotate
100         self.annotation_style = annotation_style
101         self.strip_extras = strip_extras
102         self.generate_hashes = generate_hashes
103         self.default_index_url = default_index_url
104         self.index_urls = index_urls
105         self.trusted_hosts = trusted_hosts
106         self.format_control = format_control
107         self.allow_unsafe = allow_unsafe
108         self.find_links = find_links
109         self.emit_find_links = emit_find_links
110         self.emit_options = emit_options
111 
112     def _sort_key(self, ireq: InstallRequirement) -> Tuple[bool, str]:
113         return (not ireq.editable, key_from_ireq(ireq))
114 
115     def write_header(self) -> Iterator[str]:
116         if self.emit_header:
117             yield comment("#")
118             yield comment(
119                 "# This file is autogenerated by pip-compile with python "
120                 f"{sys.version_info.major}.{sys.version_info.minor}"
121             )
122             yield comment("# To update, run:")
123             yield comment("#")
124             compile_command = os.environ.get(
125                 "CUSTOM_COMPILE_COMMAND"
126             ) or get_compile_command(self.click_ctx)
127             yield comment(f"#    {compile_command}")
128             yield comment("#")
129 
130     def write_index_options(self) -> Iterator[str]:
131         if self.emit_index_url:
132             for index, index_url in enumerate(dedup(self.index_urls)):
133                 if index == 0 and index_url.rstrip("/") == self.default_index_url:
134                     continue
135                 flag = "--index-url" if index == 0 else "--extra-index-url"
136                 yield f"{flag} {index_url}"
137 
138     def write_trusted_hosts(self) -> Iterator[str]:
139         if self.emit_trusted_host:
140             for trusted_host in dedup(self.trusted_hosts):
141                 yield f"--trusted-host {trusted_host}"
142 
143     def write_format_controls(self) -> Iterator[str]:
144         for nb in dedup(sorted(self.format_control.no_binary)):
145             yield f"--no-binary {nb}"
146         for ob in dedup(sorted(self.format_control.only_binary)):
147             yield f"--only-binary {ob}"
148 
149     def write_find_links(self) -> Iterator[str]:
150         if self.emit_find_links:
151             for find_link in dedup(self.find_links):
152                 yield f"--find-links {find_link}"
153 
154     def write_flags(self) -> Iterator[str]:
155         if not self.emit_options:
156             return
157         emitted = False
158         for line in chain(
159             self.write_index_options(),
160             self.write_find_links(),
161             self.write_trusted_hosts(),
162             self.write_format_controls(),
163         ):
164             emitted = True
165             yield line
166         if emitted:
167             yield ""
168 
169     def _iter_lines(
170         self,
171         results: Set[InstallRequirement],
172         unsafe_requirements: Optional[Set[InstallRequirement]] = None,
173         markers: Optional[Dict[str, Marker]] = None,
174         hashes: Optional[Dict[InstallRequirement, Set[str]]] = None,
175     ) -> Iterator[str]:
176         # default values
177         unsafe_requirements = unsafe_requirements or set()
178         markers = markers or {}
179         hashes = hashes or {}
180 
181         # Check for unhashed or unpinned packages if at least one package does have
182         # hashes, which will trigger pip install's --require-hashes mode.
183         warn_uninstallable = False
184         has_hashes = hashes and any(hash for hash in hashes.values())
185 
186         yielded = False
187 
188         for line in self.write_header():
189             yield line
190             yielded = True
191         for line in self.write_flags():
192             yield line
193             yielded = True
194 
195         unsafe_requirements = (
196             {r for r in results if r.name in UNSAFE_PACKAGES}
197             if not unsafe_requirements
198             else unsafe_requirements
199         )
200         packages = {r for r in results if r.name not in UNSAFE_PACKAGES}
201 
202         if packages:
203             for ireq in sorted(packages, key=self._sort_key):
204                 if has_hashes and not hashes.get(ireq):
205                     yield MESSAGE_UNHASHED_PACKAGE
206                     warn_uninstallable = True
207                 line = self._format_requirement(
208                     ireq, markers.get(key_from_ireq(ireq)), hashes=hashes
209                 )
210                 yield line
211             yielded = True
212 
213         if unsafe_requirements:
214             yield ""
215             yielded = True
216             if has_hashes and not self.allow_unsafe:
217                 yield MESSAGE_UNSAFE_PACKAGES_UNPINNED
218                 warn_uninstallable = True
219             else:
220                 yield MESSAGE_UNSAFE_PACKAGES
221 
222             for ireq in sorted(unsafe_requirements, key=self._sort_key):
223                 ireq_key = key_from_ireq(ireq)
224                 if not self.allow_unsafe:
225                     yield comment(f"# {ireq_key}")
226                 else:
227                     line = self._format_requirement(
228                         ireq, marker=markers.get(ireq_key), hashes=hashes
229                     )
230                     yield line
231 
232         # Yield even when there's no real content, so that blank files are written
233         if not yielded:
234             yield ""
235 
236         if warn_uninstallable:
237             log.warning(MESSAGE_UNINSTALLABLE)
238 
239     def write(
240         self,
241         results: Set[InstallRequirement],
242         unsafe_requirements: Set[InstallRequirement],
243         markers: Dict[str, Marker],
244         hashes: Optional[Dict[InstallRequirement, Set[str]]],
245     ) -> None:
246 
247         for line in self._iter_lines(results, unsafe_requirements, markers, hashes):
248             log.info(line)
249             if not self.dry_run:
250                 self.dst_file.write(unstyle(line).encode())
251                 self.dst_file.write(os.linesep.encode())
252 
253     def _format_requirement(
254         self,
255         ireq: InstallRequirement,
256         marker: Optional[Marker] = None,
257         hashes: Optional[Dict[InstallRequirement, Set[str]]] = None,
258     ) -> str:
259         ireq_hashes = (hashes if hashes is not None else {}).get(ireq)
260 
261         line = format_requirement(ireq, marker=marker, hashes=ireq_hashes)
262         if self.strip_extras:
263             line = re.sub(r"\[.+?\]", "", line)
264 
265         if not self.annotate:
266             return line
267 
268         # Annotate what packages or reqs-ins this package is required by
269         required_by = set()
270         if hasattr(ireq, "_source_ireqs"):
271             required_by |= {
272                 _comes_from_as_string(src_ireq)
273                 for src_ireq in ireq._source_ireqs
274                 if src_ireq.comes_from
275             }
276         elif ireq.comes_from:
277             required_by.add(_comes_from_as_string(ireq))
278 
279         if required_by:
280             if self.annotation_style == "split":
281                 annotation = annotation_style_split(required_by)
282                 sep = "\n    "
283             elif self.annotation_style == "line":
284                 annotation = annotation_style_line(required_by)
285                 sep = "\n    " if ireq_hashes else "  "
286             else:  # pragma: no cover
287                 raise ValueError("Invalid value for annotation style")
288             # 24 is one reasonable column size to use here, that we've used in the past
289             lines = f"{line:24}{sep}{comment(annotation)}".splitlines()
290             line = "\n".join(ln.rstrip() for ln in lines)
291 
292         return line
293
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/piptools/writer.py b/piptools/writer.py
--- a/piptools/writer.py
+++ b/piptools/writer.py
@@ -245,8 +245,11 @@
     ) -> None:
 
         for line in self._iter_lines(results, unsafe_requirements, markers, hashes):
-            log.info(line)
-            if not self.dry_run:
+            if self.dry_run:
+                # Bypass the log level to always print this during a dry run
+                log.log(line)
+            else:
+                log.info(line)
                 self.dst_file.write(unstyle(line).encode())
                 self.dst_file.write(os.linesep.encode())

{"golden_diff": "diff --git a/piptools/writer.py b/piptools/writer.py\n--- a/piptools/writer.py\n+++ b/piptools/writer.py\n@@ -245,8 +245,11 @@\n     ) -> None:\n \n         for line in self._iter_lines(results, unsafe_requirements, markers, hashes):\n-            log.info(line)\n-            if not self.dry_run:\n+            if self.dry_run:\n+                # Bypass the log level to always print this during a dry run\n+                log.log(line)\n+            else:\n+                log.info(line)\n                 self.dst_file.write(unstyle(line).encode())\n                 self.dst_file.write(os.linesep.encode())\n", "issue": "Incorrect output on pip-compile upgrade with --dry-run and --quiet flags\n##### Environment Versions\r\n\r\n1. OS Type: any\r\n1. Python version: any\r\n1. pip version:  any\r\n1. pip-tools version: 3.8.0\r\n\r\n##### Steps to replicate\r\n\r\n1. `echo \"six\" > requirements.in`\r\n1. `pip-compile`\r\n1. `pip-compile --upgrade --dry-run --quiet`\r\n\r\n##### Expected result\r\n\r\n- `--dry-run` should print changes to stdout/stderr\r\n- `--quiet` should suppress _\"Dry-run, so nothing updated.\"_ message \r\n\r\n```bash\r\n#\r\n# This file is autogenerated by pip-compile\r\n# To update, run:\r\n#\r\n#    pip-compile\r\n#\r\nsix==1.12.0\r\n```\r\n\r\n##### Actual result\r\n\r\nBut currently it behaves the opposite way:\r\n\r\n```\r\nDry-run, so nothing updated.\r\n```\r\n\r\n---\r\n\r\nRefs: \r\n\r\n- #843 \r\n- https://github.com/jazzband/pip-tools/issues/841#issuecomment-508454216\n", "before_files": [{"content": "import os\nimport re\nimport sys\nfrom itertools import chain\nfrom typing import BinaryIO, Dict, Iterable, Iterator, List, Optional, Set, Tuple\n\nfrom click import unstyle\nfrom click.core import Context\nfrom pip._internal.models.format_control import FormatControl\nfrom pip._internal.req.req_install import InstallRequirement\nfrom pip._vendor.packaging.markers import Marker\n\nfrom .logging import log\nfrom .utils import (\n    UNSAFE_PACKAGES,\n    comment,\n    dedup,\n    format_requirement,\n    get_compile_command,\n    key_from_ireq,\n)\n\nMESSAGE_UNHASHED_PACKAGE = comment(\n    \"# WARNING: pip install will require the following package to be hashed.\"\n    \"\\n# Consider using a hashable URL like \"\n    \"https://github.com/jazzband/pip-tools/archive/SOMECOMMIT.zip\"\n)\n\nMESSAGE_UNSAFE_PACKAGES_UNPINNED = comment(\n    \"# WARNING: The following packages were not pinned, but pip requires them to be\"\n    \"\\n# pinned when the requirements file includes hashes. \"\n    \"Consider using the --allow-unsafe flag.\"\n)\n\nMESSAGE_UNSAFE_PACKAGES = comment(\n    \"# The following packages are considered to be unsafe in a requirements file:\"\n)\n\nMESSAGE_UNINSTALLABLE = (\n    \"The generated requirements file may be rejected by pip install. \"\n    \"See # WARNING lines for details.\"\n)\n\n\nstrip_comes_from_line_re = re.compile(r\" \\(line \\d+\\)$\")\n\n\ndef _comes_from_as_string(ireq: InstallRequirement) -> str:\n    if isinstance(ireq.comes_from, str):\n        return strip_comes_from_line_re.sub(\"\", ireq.comes_from)\n    return key_from_ireq(ireq.comes_from)\n\n\ndef annotation_style_split(required_by: Set[str]) -> str:\n    sorted_required_by = sorted(required_by)\n    if len(sorted_required_by) == 1:\n        source = sorted_required_by[0]\n        annotation = \"# via \" + source\n    else:\n        annotation_lines = [\"# via\"]\n        for source in sorted_required_by:\n            annotation_lines.append(\"    #   \" + source)\n        annotation = \"\\n\".join(annotation_lines)\n    return annotation\n\n\ndef annotation_style_line(required_by: Set[str]) -> str:\n    return f\"# via {', '.join(sorted(required_by))}\"\n\n\nclass OutputWriter:\n    def __init__(\n        self,\n        dst_file: BinaryIO,\n        click_ctx: Context,\n        dry_run: bool,\n        emit_header: bool,\n        emit_index_url: bool,\n        emit_trusted_host: bool,\n        annotate: bool,\n        annotation_style: str,\n        strip_extras: bool,\n        generate_hashes: bool,\n        default_index_url: str,\n        index_urls: Iterable[str],\n        trusted_hosts: Iterable[str],\n        format_control: FormatControl,\n        allow_unsafe: bool,\n        find_links: List[str],\n        emit_find_links: bool,\n        emit_options: bool,\n    ) -> None:\n        self.dst_file = dst_file\n        self.click_ctx = click_ctx\n        self.dry_run = dry_run\n        self.emit_header = emit_header\n        self.emit_index_url = emit_index_url\n        self.emit_trusted_host = emit_trusted_host\n        self.annotate = annotate\n        self.annotation_style = annotation_style\n        self.strip_extras = strip_extras\n        self.generate_hashes = generate_hashes\n        self.default_index_url = default_index_url\n        self.index_urls = index_urls\n        self.trusted_hosts = trusted_hosts\n        self.format_control = format_control\n        self.allow_unsafe = allow_unsafe\n        self.find_links = find_links\n        self.emit_find_links = emit_find_links\n        self.emit_options = emit_options\n\n    def _sort_key(self, ireq: InstallRequirement) -> Tuple[bool, str]:\n        return (not ireq.editable, key_from_ireq(ireq))\n\n    def write_header(self) -> Iterator[str]:\n        if self.emit_header:\n            yield comment(\"#\")\n            yield comment(\n                \"# This file is autogenerated by pip-compile with python \"\n                f\"{sys.version_info.major}.{sys.version_info.minor}\"\n            )\n            yield comment(\"# To update, run:\")\n            yield comment(\"#\")\n            compile_command = os.environ.get(\n                \"CUSTOM_COMPILE_COMMAND\"\n            ) or get_compile_command(self.click_ctx)\n            yield comment(f\"#    {compile_command}\")\n            yield comment(\"#\")\n\n    def write_index_options(self) -> Iterator[str]:\n        if self.emit_index_url:\n            for index, index_url in enumerate(dedup(self.index_urls)):\n                if index == 0 and index_url.rstrip(\"/\") == self.default_index_url:\n                    continue\n                flag = \"--index-url\" if index == 0 else \"--extra-index-url\"\n                yield f\"{flag} {index_url}\"\n\n    def write_trusted_hosts(self) -> Iterator[str]:\n        if self.emit_trusted_host:\n            for trusted_host in dedup(self.trusted_hosts):\n                yield f\"--trusted-host {trusted_host}\"\n\n    def write_format_controls(self) -> Iterator[str]:\n        for nb in dedup(sorted(self.format_control.no_binary)):\n            yield f\"--no-binary {nb}\"\n        for ob in dedup(sorted(self.format_control.only_binary)):\n            yield f\"--only-binary {ob}\"\n\n    def write_find_links(self) -> Iterator[str]:\n        if self.emit_find_links:\n            for find_link in dedup(self.find_links):\n                yield f\"--find-links {find_link}\"\n\n    def write_flags(self) -> Iterator[str]:\n        if not self.emit_options:\n            return\n        emitted = False\n        for line in chain(\n            self.write_index_options(),\n            self.write_find_links(),\n            self.write_trusted_hosts(),\n            self.write_format_controls(),\n        ):\n            emitted = True\n            yield line\n        if emitted:\n            yield \"\"\n\n    def _iter_lines(\n        self,\n        results: Set[InstallRequirement],\n        unsafe_requirements: Optional[Set[InstallRequirement]] = None,\n        markers: Optional[Dict[str, Marker]] = None,\n        hashes: Optional[Dict[InstallRequirement, Set[str]]] = None,\n    ) -> Iterator[str]:\n        # default values\n        unsafe_requirements = unsafe_requirements or set()\n        markers = markers or {}\n        hashes = hashes or {}\n\n        # Check for unhashed or unpinned packages if at least one package does have\n        # hashes, which will trigger pip install's --require-hashes mode.\n        warn_uninstallable = False\n        has_hashes = hashes and any(hash for hash in hashes.values())\n\n        yielded = False\n\n        for line in self.write_header():\n            yield line\n            yielded = True\n        for line in self.write_flags():\n            yield line\n            yielded = True\n\n        unsafe_requirements = (\n            {r for r in results if r.name in UNSAFE_PACKAGES}\n            if not unsafe_requirements\n            else unsafe_requirements\n        )\n        packages = {r for r in results if r.name not in UNSAFE_PACKAGES}\n\n        if packages:\n            for ireq in sorted(packages, key=self._sort_key):\n                if has_hashes and not hashes.get(ireq):\n                    yield MESSAGE_UNHASHED_PACKAGE\n                    warn_uninstallable = True\n                line = self._format_requirement(\n                    ireq, markers.get(key_from_ireq(ireq)), hashes=hashes\n                )\n                yield line\n            yielded = True\n\n        if unsafe_requirements:\n            yield \"\"\n            yielded = True\n            if has_hashes and not self.allow_unsafe:\n                yield MESSAGE_UNSAFE_PACKAGES_UNPINNED\n                warn_uninstallable = True\n            else:\n                yield MESSAGE_UNSAFE_PACKAGES\n\n            for ireq in sorted(unsafe_requirements, key=self._sort_key):\n                ireq_key = key_from_ireq(ireq)\n                if not self.allow_unsafe:\n                    yield comment(f\"# {ireq_key}\")\n                else:\n                    line = self._format_requirement(\n                        ireq, marker=markers.get(ireq_key), hashes=hashes\n                    )\n                    yield line\n\n        # Yield even when there's no real content, so that blank files are written\n        if not yielded:\n            yield \"\"\n\n        if warn_uninstallable:\n            log.warning(MESSAGE_UNINSTALLABLE)\n\n    def write(\n        self,\n        results: Set[InstallRequirement],\n        unsafe_requirements: Set[InstallRequirement],\n        markers: Dict[str, Marker],\n        hashes: Optional[Dict[InstallRequirement, Set[str]]],\n    ) -> None:\n\n        for line in self._iter_lines(results, unsafe_requirements, markers, hashes):\n            log.info(line)\n            if not self.dry_run:\n                self.dst_file.write(unstyle(line).encode())\n                self.dst_file.write(os.linesep.encode())\n\n    def _format_requirement(\n        self,\n        ireq: InstallRequirement,\n        marker: Optional[Marker] = None,\n        hashes: Optional[Dict[InstallRequirement, Set[str]]] = None,\n    ) -> str:\n        ireq_hashes = (hashes if hashes is not None else {}).get(ireq)\n\n        line = format_requirement(ireq, marker=marker, hashes=ireq_hashes)\n        if self.strip_extras:\n            line = re.sub(r\"\\[.+?\\]\", \"\", line)\n\n        if not self.annotate:\n            return line\n\n        # Annotate what packages or reqs-ins this package is required by\n        required_by = set()\n        if hasattr(ireq, \"_source_ireqs\"):\n            required_by |= {\n                _comes_from_as_string(src_ireq)\n                for src_ireq in ireq._source_ireqs\n                if src_ireq.comes_from\n            }\n        elif ireq.comes_from:\n            required_by.add(_comes_from_as_string(ireq))\n\n        if required_by:\n            if self.annotation_style == \"split\":\n                annotation = annotation_style_split(required_by)\n                sep = \"\\n    \"\n            elif self.annotation_style == \"line\":\n                annotation = annotation_style_line(required_by)\n                sep = \"\\n    \" if ireq_hashes else \"  \"\n            else:  # pragma: no cover\n                raise ValueError(\"Invalid value for annotation style\")\n            # 24 is one reasonable column size to use here, that we've used in the past\n            lines = f\"{line:24}{sep}{comment(annotation)}\".splitlines()\n            line = \"\\n\".join(ln.rstrip() for ln in lines)\n\n        return line\n", "path": "piptools/writer.py"}], "after_files": [{"content": "import os\nimport re\nimport sys\nfrom itertools import chain\nfrom typing import BinaryIO, Dict, Iterable, Iterator, List, Optional, Set, Tuple\n\nfrom click import unstyle\nfrom click.core import Context\nfrom pip._internal.models.format_control import FormatControl\nfrom pip._internal.req.req_install import InstallRequirement\nfrom pip._vendor.packaging.markers import Marker\n\nfrom .logging import log\nfrom .utils import (\n    UNSAFE_PACKAGES,\n    comment,\n    dedup,\n    format_requirement,\n    get_compile_command,\n    key_from_ireq,\n)\n\nMESSAGE_UNHASHED_PACKAGE = comment(\n    \"# WARNING: pip install will require the following package to be hashed.\"\n    \"\\n# Consider using a hashable URL like \"\n    \"https://github.com/jazzband/pip-tools/archive/SOMECOMMIT.zip\"\n)\n\nMESSAGE_UNSAFE_PACKAGES_UNPINNED = comment(\n    \"# WARNING: The following packages were not pinned, but pip requires them to be\"\n    \"\\n# pinned when the requirements file includes hashes. \"\n    \"Consider using the --allow-unsafe flag.\"\n)\n\nMESSAGE_UNSAFE_PACKAGES = comment(\n    \"# The following packages are considered to be unsafe in a requirements file:\"\n)\n\nMESSAGE_UNINSTALLABLE = (\n    \"The generated requirements file may be rejected by pip install. \"\n    \"See # WARNING lines for details.\"\n)\n\n\nstrip_comes_from_line_re = re.compile(r\" \\(line \\d+\\)$\")\n\n\ndef _comes_from_as_string(ireq: InstallRequirement) -> str:\n    if isinstance(ireq.comes_from, str):\n        return strip_comes_from_line_re.sub(\"\", ireq.comes_from)\n    return key_from_ireq(ireq.comes_from)\n\n\ndef annotation_style_split(required_by: Set[str]) -> str:\n    sorted_required_by = sorted(required_by)\n    if len(sorted_required_by) == 1:\n        source = sorted_required_by[0]\n        annotation = \"# via \" + source\n    else:\n        annotation_lines = [\"# via\"]\n        for source in sorted_required_by:\n            annotation_lines.append(\"    #   \" + source)\n        annotation = \"\\n\".join(annotation_lines)\n    return annotation\n\n\ndef annotation_style_line(required_by: Set[str]) -> str:\n    return f\"# via {', '.join(sorted(required_by))}\"\n\n\nclass OutputWriter:\n    def __init__(\n        self,\n        dst_file: BinaryIO,\n        click_ctx: Context,\n        dry_run: bool,\n        emit_header: bool,\n        emit_index_url: bool,\n        emit_trusted_host: bool,\n        annotate: bool,\n        annotation_style: str,\n        strip_extras: bool,\n        generate_hashes: bool,\n        default_index_url: str,\n        index_urls: Iterable[str],\n        trusted_hosts: Iterable[str],\n        format_control: FormatControl,\n        allow_unsafe: bool,\n        find_links: List[str],\n        emit_find_links: bool,\n        emit_options: bool,\n    ) -> None:\n        self.dst_file = dst_file\n        self.click_ctx = click_ctx\n        self.dry_run = dry_run\n        self.emit_header = emit_header\n        self.emit_index_url = emit_index_url\n        self.emit_trusted_host = emit_trusted_host\n        self.annotate = annotate\n        self.annotation_style = annotation_style\n        self.strip_extras = strip_extras\n        self.generate_hashes = generate_hashes\n        self.default_index_url = default_index_url\n        self.index_urls = index_urls\n        self.trusted_hosts = trusted_hosts\n        self.format_control = format_control\n        self.allow_unsafe = allow_unsafe\n        self.find_links = find_links\n        self.emit_find_links = emit_find_links\n        self.emit_options = emit_options\n\n    def _sort_key(self, ireq: InstallRequirement) -> Tuple[bool, str]:\n        return (not ireq.editable, key_from_ireq(ireq))\n\n    def write_header(self) -> Iterator[str]:\n        if self.emit_header:\n            yield comment(\"#\")\n            yield comment(\n                \"# This file is autogenerated by pip-compile with python \"\n                f\"{sys.version_info.major}.{sys.version_info.minor}\"\n            )\n            yield comment(\"# To update, run:\")\n            yield comment(\"#\")\n            compile_command = os.environ.get(\n                \"CUSTOM_COMPILE_COMMAND\"\n            ) or get_compile_command(self.click_ctx)\n            yield comment(f\"#    {compile_command}\")\n            yield comment(\"#\")\n\n    def write_index_options(self) -> Iterator[str]:\n        if self.emit_index_url:\n            for index, index_url in enumerate(dedup(self.index_urls)):\n                if index == 0 and index_url.rstrip(\"/\") == self.default_index_url:\n                    continue\n                flag = \"--index-url\" if index == 0 else \"--extra-index-url\"\n                yield f\"{flag} {index_url}\"\n\n    def write_trusted_hosts(self) -> Iterator[str]:\n        if self.emit_trusted_host:\n            for trusted_host in dedup(self.trusted_hosts):\n                yield f\"--trusted-host {trusted_host}\"\n\n    def write_format_controls(self) -> Iterator[str]:\n        for nb in dedup(sorted(self.format_control.no_binary)):\n            yield f\"--no-binary {nb}\"\n        for ob in dedup(sorted(self.format_control.only_binary)):\n            yield f\"--only-binary {ob}\"\n\n    def write_find_links(self) -> Iterator[str]:\n        if self.emit_find_links:\n            for find_link in dedup(self.find_links):\n                yield f\"--find-links {find_link}\"\n\n    def write_flags(self) -> Iterator[str]:\n        if not self.emit_options:\n            return\n        emitted = False\n        for line in chain(\n            self.write_index_options(),\n            self.write_find_links(),\n            self.write_trusted_hosts(),\n            self.write_format_controls(),\n        ):\n            emitted = True\n            yield line\n        if emitted:\n            yield \"\"\n\n    def _iter_lines(\n        self,\n        results: Set[InstallRequirement],\n        unsafe_requirements: Optional[Set[InstallRequirement]] = None,\n        markers: Optional[Dict[str, Marker]] = None,\n        hashes: Optional[Dict[InstallRequirement, Set[str]]] = None,\n    ) -> Iterator[str]:\n        # default values\n        unsafe_requirements = unsafe_requirements or set()\n        markers = markers or {}\n        hashes = hashes or {}\n\n        # Check for unhashed or unpinned packages if at least one package does have\n        # hashes, which will trigger pip install's --require-hashes mode.\n        warn_uninstallable = False\n        has_hashes = hashes and any(hash for hash in hashes.values())\n\n        yielded = False\n\n        for line in self.write_header():\n            yield line\n            yielded = True\n        for line in self.write_flags():\n            yield line\n            yielded = True\n\n        unsafe_requirements = (\n            {r for r in results if r.name in UNSAFE_PACKAGES}\n            if not unsafe_requirements\n            else unsafe_requirements\n        )\n        packages = {r for r in results if r.name not in UNSAFE_PACKAGES}\n\n        if packages:\n            for ireq in sorted(packages, key=self._sort_key):\n                if has_hashes and not hashes.get(ireq):\n                    yield MESSAGE_UNHASHED_PACKAGE\n                    warn_uninstallable = True\n                line = self._format_requirement(\n                    ireq, markers.get(key_from_ireq(ireq)), hashes=hashes\n                )\n                yield line\n            yielded = True\n\n        if unsafe_requirements:\n            yield \"\"\n            yielded = True\n            if has_hashes and not self.allow_unsafe:\n                yield MESSAGE_UNSAFE_PACKAGES_UNPINNED\n                warn_uninstallable = True\n            else:\n                yield MESSAGE_UNSAFE_PACKAGES\n\n            for ireq in sorted(unsafe_requirements, key=self._sort_key):\n                ireq_key = key_from_ireq(ireq)\n                if not self.allow_unsafe:\n                    yield comment(f\"# {ireq_key}\")\n                else:\n                    line = self._format_requirement(\n                        ireq, marker=markers.get(ireq_key), hashes=hashes\n                    )\n                    yield line\n\n        # Yield even when there's no real content, so that blank files are written\n        if not yielded:\n            yield \"\"\n\n        if warn_uninstallable:\n            log.warning(MESSAGE_UNINSTALLABLE)\n\n    def write(\n        self,\n        results: Set[InstallRequirement],\n        unsafe_requirements: Set[InstallRequirement],\n        markers: Dict[str, Marker],\n        hashes: Optional[Dict[InstallRequirement, Set[str]]],\n    ) -> None:\n\n        for line in self._iter_lines(results, unsafe_requirements, markers, hashes):\n            if self.dry_run:\n                # Bypass the log level to always print this during a dry run\n                log.log(line)\n            else:\n                log.info(line)\n                self.dst_file.write(unstyle(line).encode())\n                self.dst_file.write(os.linesep.encode())\n\n    def _format_requirement(\n        self,\n        ireq: InstallRequirement,\n        marker: Optional[Marker] = None,\n        hashes: Optional[Dict[InstallRequirement, Set[str]]] = None,\n    ) -> str:\n        ireq_hashes = (hashes if hashes is not None else {}).get(ireq)\n\n        line = format_requirement(ireq, marker=marker, hashes=ireq_hashes)\n        if self.strip_extras:\n            line = re.sub(r\"\\[.+?\\]\", \"\", line)\n\n        if not self.annotate:\n            return line\n\n        # Annotate what packages or reqs-ins this package is required by\n        required_by = set()\n        if hasattr(ireq, \"_source_ireqs\"):\n            required_by |= {\n                _comes_from_as_string(src_ireq)\n                for src_ireq in ireq._source_ireqs\n                if src_ireq.comes_from\n            }\n        elif ireq.comes_from:\n            required_by.add(_comes_from_as_string(ireq))\n\n        if required_by:\n            if self.annotation_style == \"split\":\n                annotation = annotation_style_split(required_by)\n                sep = \"\\n    \"\n            elif self.annotation_style == \"line\":\n                annotation = annotation_style_line(required_by)\n                sep = \"\\n    \" if ireq_hashes else \"  \"\n            else:  # pragma: no cover\n                raise ValueError(\"Invalid value for annotation style\")\n            # 24 is one reasonable column size to use here, that we've used in the past\n            lines = f\"{line:24}{sep}{comment(annotation)}\".splitlines()\n            line = \"\\n\".join(ln.rstrip() for ln in lines)\n\n        return line\n", "path": "piptools/writer.py"}]}