We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Improve --allow-unsafe warning comment
Currently, if `pip-compile` suppresses a requirement because the `--allow-unsafe` flag wasn't passed, it includes a warning that reads something like this:

```
# WARNING: The following packages were not pinned, but pip requires them to be
# pinned when the requirements file includes hashes. Consider using the --allow-unsafe flag.
# pip
```

This wording is a little incorrect because it's not always the case that pip requires them to be pinned. Namely, if the requirement is already satisfied by a package that's already installed, then pip won't check any hashes (because it won't need to install anything for that requirement). Because of this, a user seeing the current message can get the mistaken idea that they don't have any options other than passing `--allow-unsafe`.

Thus, I'd like to suggest that the wording be modified slightly to read something like:

```
# WARNING: The following packages were not pinned, but pip requires them to be
# pinned when the requirements file includes hashes and the requirement isn't
# already satisfied by a package already installed. Consider using the
# --allow-unsafe flag.
# pip
```

(As a side note, there is a third option in addition to (1) commenting out the "unsafe" requirement, and (2) including the pinned requirement with hashes. Namely, the pinned requirement can be included without hashes. Doing that third option would enforce that the user has the correct version already installed by erroring out if they don't, while not risking doing an "unsafe" operation by uninstalling / installing the package in question. The reason is that, by not including the hashes, it prevents a new install from taking place.)

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

{"golden_diff": "diff --git a/piptools/writer.py b/piptools/writer.py\n--- a/piptools/writer.py\n+++ b/piptools/writer.py\n@@ -32,7 +32,8 @@\n \n MESSAGE_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+    \"\\n# pinned when the requirements file includes hashes and the requirement is not\"\n+    \"\\n# satisfied by a package already installed. \"\n     \"Consider using the --allow-unsafe flag.\"\n )\n", "issue": "Improve --allow-unsafe warning comment\nCurrently, if `pip-compile` suppresses a requirement because the `--allow-unsafe` flag wasn't passed, it includes a warning that reads something like this:\r\n\r\n```\r\n# WARNING: The following packages were not pinned, but pip requires them to be\r\n# pinned when the requirements file includes hashes. Consider using the --allow-unsafe flag.\r\n# pip\r\n```\r\n\r\nThis wording is a little incorrect because it's not always the case that pip requires them to be pinned. Namely, if the requirement is already satisfied by a package that's already installed, then pip won't check any hashes (because it won't need to install anything for that requirement). Because of this, a user seeing the current message can get the mistaken idea that they don't have any options other than passing `--allow-unsafe`.\r\n\r\nThus, I'd like to suggest that the wording be modified slightly to read something like:\r\n\r\n```\r\n# WARNING: The following packages were not pinned, but pip requires them to be\r\n# pinned when the requirements file includes hashes and the requirement isn't\r\n# already satisfied by a package already installed. Consider using the\r\n# --allow-unsafe flag.\r\n# pip\r\n```\r\n\r\n(As a side note, there is a third option in addition to (1) commenting out the \"unsafe\" requirement, and (2) including the pinned requirement with hashes. Namely, the pinned requirement can be included without hashes. Doing that third option would enforce that the user has the correct version already installed by erroring out if they don't, while not risking doing an \"unsafe\" operation by uninstalling / installing the package in question. The reason is that, by not including the hashes, it prevents a new install from taking place.)\n", "before_files": [{"content": "from __future__ import annotations\n\nimport io\nimport os\nimport re\nimport sys\nfrom itertools import chain\nfrom typing import BinaryIO, Iterable, Iterator, cast\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\nfrom pip._vendor.packaging.utils import canonicalize_name\n\nfrom .logging import log\nfrom .utils import (\n    comment,\n    dedup,\n    format_requirement,\n    get_compile_command,\n    key_from_ireq,\n    strip_extras,\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(comes_from: str | InstallRequirement) -> str:\n    if isinstance(comes_from, str):\n        return strip_comes_from_line_re.sub(\"\", comes_from)\n    return cast(str, canonicalize_name(key_from_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        linesep: str,\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.linesep = linesep\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(\"# by the following command:\")\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: set[InstallRequirement],\n        unsafe_packages: set[str],\n        markers: dict[str, Marker],\n        hashes: dict[InstallRequirement, set[str]] | None = None,\n    ) -> Iterator[str]:\n        # default values\n        unsafe_packages = unsafe_packages if self.allow_unsafe else set()\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 = unsafe_requirements or {\n            r for r in results if r.name in unsafe_packages\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        unsafe_packages: set[str],\n        markers: dict[str, Marker],\n        hashes: dict[InstallRequirement, set[str]] | None,\n    ) -> None:\n        if not self.dry_run:\n            dst_file = io.TextIOWrapper(\n                self.dst_file,\n                encoding=\"utf8\",\n                newline=self.linesep,\n                line_buffering=True,\n            )\n        try:\n            for line in self._iter_lines(\n                results, unsafe_requirements, unsafe_packages, markers, hashes\n            ):\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                    dst_file.write(unstyle(line))\n                    dst_file.write(\"\\n\")\n        finally:\n            if not self.dry_run:\n                dst_file.detach()\n\n    def _format_requirement(\n        self,\n        ireq: InstallRequirement,\n        marker: Marker | None = None,\n        hashes: dict[InstallRequirement, set[str]] | None = 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 = strip_extras(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.comes_from)\n                for src_ireq in ireq._source_ireqs\n                if src_ireq.comes_from\n            }\n\n        if ireq.comes_from:\n            required_by.add(_comes_from_as_string(ireq.comes_from))\n\n        required_by |= set(getattr(ireq, \"_required_by\", set()))\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            if self.strip_extras:\n                annotation = strip_extras(annotation)\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": "from __future__ import annotations\n\nimport io\nimport os\nimport re\nimport sys\nfrom itertools import chain\nfrom typing import BinaryIO, Iterable, Iterator, cast\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\nfrom pip._vendor.packaging.utils import canonicalize_name\n\nfrom .logging import log\nfrom .utils import (\n    comment,\n    dedup,\n    format_requirement,\n    get_compile_command,\n    key_from_ireq,\n    strip_extras,\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 and the requirement is not\"\n    \"\\n# satisfied by a package already installed. \"\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(comes_from: str | InstallRequirement) -> str:\n    if isinstance(comes_from, str):\n        return strip_comes_from_line_re.sub(\"\", comes_from)\n    return cast(str, canonicalize_name(key_from_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        linesep: str,\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.linesep = linesep\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(\"# by the following command:\")\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: set[InstallRequirement],\n        unsafe_packages: set[str],\n        markers: dict[str, Marker],\n        hashes: dict[InstallRequirement, set[str]] | None = None,\n    ) -> Iterator[str]:\n        # default values\n        unsafe_packages = unsafe_packages if self.allow_unsafe else set()\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 = unsafe_requirements or {\n            r for r in results if r.name in unsafe_packages\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        unsafe_packages: set[str],\n        markers: dict[str, Marker],\n        hashes: dict[InstallRequirement, set[str]] | None,\n    ) -> None:\n        if not self.dry_run:\n            dst_file = io.TextIOWrapper(\n                self.dst_file,\n                encoding=\"utf8\",\n                newline=self.linesep,\n                line_buffering=True,\n            )\n        try:\n            for line in self._iter_lines(\n                results, unsafe_requirements, unsafe_packages, markers, hashes\n            ):\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                    dst_file.write(unstyle(line))\n                    dst_file.write(\"\\n\")\n        finally:\n            if not self.dry_run:\n                dst_file.detach()\n\n    def _format_requirement(\n        self,\n        ireq: InstallRequirement,\n        marker: Marker | None = None,\n        hashes: dict[InstallRequirement, set[str]] | None = 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 = strip_extras(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.comes_from)\n                for src_ireq in ireq._source_ireqs\n                if src_ireq.comes_from\n            }\n\n        if ireq.comes_from:\n            required_by.add(_comes_from_as_string(ireq.comes_from))\n\n        required_by |= set(getattr(ireq, \"_required_by\", set()))\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            if self.strip_extras:\n                annotation = strip_extras(annotation)\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"}]}