We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Output --no-binary and --only-binary options to preserve pip behavior
<!-- Describe the issue briefly here. -->

#### Environment Versions

1. OS Type: Linux
1. Python version: 3.12.3
1. pip version: 24.0
1. pip-tools version: 7.4.1

#### Steps to replicate

1. Compile using `--pip-args='--only-binary=:all: --no-binary=library'
2. See that compiled requirements list the `--no-binary=library` option first, `--only-binary=:all:` option second.
3. When attempting to install from these requirements the `--no-binary` is wiped out by the `--only-binary=:all:`

#### Expected result

The resulting requirements contain `--no-binary` and `--only-binary` options that have the same behavior as input options.

#### Actual result

Requirements don't have the same behavior as input options.

This improvement matters because using --no-binary/--only-binary is the best way for users to control the amount of potential code execution that happens during the installation process. I have a local fix that I plan on creating a PR.


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

{"golden_diff": "diff --git a/piptools/writer.py b/piptools/writer.py\n--- a/piptools/writer.py\n+++ b/piptools/writer.py\n@@ -148,9 +148,27 @@\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+        # The ordering of output needs to preserve the behavior of pip's\n+        # FormatControl.get_allowed_formats(). The behavior is the following:\n+        #\n+        #   * Parsing of CLI options happens first to last.\n+        #   * --only-binary takes precedence over --no-binary\n+        #   * Package names take precedence over :all:\n+        #   * We'll never see :all: in both due to mutual exclusion.\n+        #\n+        # So in summary, we want to emit :all: first and then package names later.\n+        no_binary = self.format_control.no_binary.copy()\n+        only_binary = self.format_control.only_binary.copy()\n+\n+        if \":all:\" in no_binary:\n+            yield \"--no-binary :all:\"\n+            no_binary.remove(\":all:\")\n+        if \":all:\" in only_binary:\n+            yield \"--only-binary :all:\"\n+            only_binary.remove(\":all:\")\n+        for nb in dedup(sorted(no_binary)):\n             yield f\"--no-binary {nb}\"\n-        for ob in dedup(sorted(self.format_control.only_binary)):\n+        for ob in dedup(sorted(only_binary)):\n             yield f\"--only-binary {ob}\"\n \n     def write_find_links(self) -> Iterator[str]:\n", "issue": "Output --no-binary and --only-binary options to preserve pip behavior\n<!-- Describe the issue briefly here. -->\r\n\r\n#### Environment Versions\r\n\r\n1. OS Type: Linux\r\n1. Python version: 3.12.3\r\n1. pip version: 24.0\r\n1. pip-tools version: 7.4.1\r\n\r\n#### Steps to replicate\r\n\r\n1. Compile using `--pip-args='--only-binary=:all: --no-binary=library'\r\n2. See that compiled requirements list the `--no-binary=library` option first, `--only-binary=:all:` option second.\r\n3. When attempting to install from these requirements the `--no-binary` is wiped out by the `--only-binary=:all:`\r\n\r\n#### Expected result\r\n\r\nThe resulting requirements contain `--no-binary` and `--only-binary` options that have the same behavior as input options.\r\n\r\n#### Actual result\r\n\r\nRequirements don't have the same behavior as input options.\r\n\r\nThis improvement matters because using --no-binary/--only-binary is the best way for users to control the amount of potential code execution that happens during the installation process. I have a local fix that I plan on creating a PR.\r\n\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 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        # Filter out the origin install requirements for extras.\n        # See https://github.com/jazzband/pip-tools/issues/2003\n        if ireq.comes_from and (\n            isinstance(ireq.comes_from, str) or ireq.comes_from.name != ireq.name\n        ):\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        # The ordering of output needs to preserve the behavior of pip's\n        # FormatControl.get_allowed_formats(). The behavior is the following:\n        #\n        #   * Parsing of CLI options happens first to last.\n        #   * --only-binary takes precedence over --no-binary\n        #   * Package names take precedence over :all:\n        #   * We'll never see :all: in both due to mutual exclusion.\n        #\n        # So in summary, we want to emit :all: first and then package names later.\n        no_binary = self.format_control.no_binary.copy()\n        only_binary = self.format_control.only_binary.copy()\n\n        if \":all:\" in no_binary:\n            yield \"--no-binary :all:\"\n            no_binary.remove(\":all:\")\n        if \":all:\" in only_binary:\n            yield \"--only-binary :all:\"\n            only_binary.remove(\":all:\")\n        for nb in dedup(sorted(no_binary)):\n            yield f\"--no-binary {nb}\"\n        for ob in dedup(sorted(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        # Filter out the origin install requirements for extras.\n        # See https://github.com/jazzband/pip-tools/issues/2003\n        if ireq.comes_from and (\n            isinstance(ireq.comes_from, str) or ireq.comes_from.name != ireq.name\n        ):\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"}]}