We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Pants 2.19 fails to download WHLs with URL-escaped chars
**Describe the bug**
Pants 2.19 fails to download WHL files hosted on URLs that contain URL-escaped characters. This applies to both URLs already present in the lockfile and to URLs retrieved during `generate-lockfiles`.

For example, instead of:
https://download.pytorch.org/whl/cpu/torch-2.0.1%2Bcpu-cp311-cp311-linux_x86_64.whl
it attempts (and fails) to download:
https://download.pytorch.org/whl/cpu/torch-2.0.1+cpu-cp311-cp311-linux_x86_64.whl

Seems to be related to the `v2.1.148` PEX version, as downgrading to `v2.1.137` resolves the issue

**Pants version**
2.19.0

**OS**
Both MacOS and Linux

**Additional info**
Minimalist example to reproduce:

```BUILD
python_requirement(
    requirements=["torch==2.0.1+cpu"],
)
```

```pants.toml
[GLOBAL]
pants_version = "2.19.0"

backend_packages = [
  "pants.backend.python",
]

[python]
interpreter_constraints = ["==3.11.*"]
enable_resolves = true

[python-repos]
find_links = [
    "https://download.pytorch.org/whl/torch_stable.html",
]

# Workaround: Downgrade to a previous PEX version
#[pex-cli]
#known_versions = [
#  "v2.1.137|macos_arm64|faad51a6a108fba9d40b2a10e82a2646fccbaf8c3d9be47818f4bffae02d94b8|4098329",
#  "v2.1.137|macos_x86_64|faad51a6a108fba9d40b2a10e82a2646fccbaf8c3d9be47818f4bffae02d94b8|4098329",
#  "v2.1.137|linux_x86_64|faad51a6a108fba9d40b2a10e82a2646fccbaf8c3d9be47818f4bffae02d94b8|4098329",
#  "v2.1.137|linux_arm64|faad51a6a108fba9d40b2a10e82a2646fccbaf8c3d9be47818f4bffae02d94b8|4098329"
#]
#version = "v2.1.137"

```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/python/pants/backend/python/util_rules/pex_cli.py`
Content:
```
1 # Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2 # Licensed under the Apache License, Version 2.0 (see LICENSE).
3 
4 from __future__ import annotations
5 
6 import dataclasses
7 import os.path
8 from dataclasses import dataclass
9 from typing import Iterable, List, Mapping, Optional, Tuple
10 
11 from pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem
12 from pants.backend.python.subsystems.setup import PythonSetup
13 from pants.backend.python.util_rules import pex_environment
14 from pants.backend.python.util_rules.pex_environment import PexEnvironment, PexSubsystem
15 from pants.core.util_rules import adhoc_binaries, external_tool
16 from pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary
17 from pants.core.util_rules.external_tool import (
18     DownloadedExternalTool,
19     ExternalToolRequest,
20     TemplatedExternalTool,
21 )
22 from pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests
23 from pants.engine.internals.selectors import MultiGet
24 from pants.engine.platform import Platform
25 from pants.engine.process import Process, ProcessCacheScope
26 from pants.engine.rules import Get, collect_rules, rule
27 from pants.option.global_options import GlobalOptions, ca_certs_path_to_file_content
28 from pants.util.frozendict import FrozenDict
29 from pants.util.logging import LogLevel
30 from pants.util.meta import classproperty
31 
32 
33 class PexCli(TemplatedExternalTool):
34     options_scope = "pex-cli"
35     name = "pex"
36     help = "The PEX (Python EXecutable) tool (https://github.com/pantsbuild/pex)."
37 
38     default_version = "v2.1.159"
39     default_url_template = "https://github.com/pantsbuild/pex/releases/download/{version}/pex"
40     version_constraints = ">=2.1.135,<3.0"
41 
42     @classproperty
43     def default_known_versions(cls):
44         return [
45             "|".join(
46                 (
47                     cls.default_version,
48                     plat,
49                     "83c3090938b4d276703864c34ba50bcb3616db0663c54b56dd0521a668d9555f",
50                     "3671772",
51                 )
52             )
53             for plat in ["macos_arm64", "macos_x86_64", "linux_x86_64", "linux_arm64"]
54         ]
55 
56 
57 @dataclass(frozen=True)
58 class PexCliProcess:
59     subcommand: tuple[str, ...]
60     extra_args: tuple[str, ...]
61     description: str = dataclasses.field(compare=False)
62     additional_input_digest: Optional[Digest]
63     extra_env: Optional[FrozenDict[str, str]]
64     output_files: Optional[Tuple[str, ...]]
65     output_directories: Optional[Tuple[str, ...]]
66     level: LogLevel
67     concurrency_available: int
68     cache_scope: ProcessCacheScope
69 
70     def __init__(
71         self,
72         *,
73         subcommand: Iterable[str],
74         extra_args: Iterable[str],
75         description: str,
76         additional_input_digest: Optional[Digest] = None,
77         extra_env: Optional[Mapping[str, str]] = None,
78         output_files: Optional[Iterable[str]] = None,
79         output_directories: Optional[Iterable[str]] = None,
80         level: LogLevel = LogLevel.INFO,
81         concurrency_available: int = 0,
82         cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,
83     ) -> None:
84         object.__setattr__(self, "subcommand", tuple(subcommand))
85         object.__setattr__(self, "extra_args", tuple(extra_args))
86         object.__setattr__(self, "description", description)
87         object.__setattr__(self, "additional_input_digest", additional_input_digest)
88         object.__setattr__(self, "extra_env", FrozenDict(extra_env) if extra_env else None)
89         object.__setattr__(self, "output_files", tuple(output_files) if output_files else None)
90         object.__setattr__(
91             self, "output_directories", tuple(output_directories) if output_directories else None
92         )
93         object.__setattr__(self, "level", level)
94         object.__setattr__(self, "concurrency_available", concurrency_available)
95         object.__setattr__(self, "cache_scope", cache_scope)
96 
97         self.__post_init__()
98 
99     def __post_init__(self) -> None:
100         if "--pex-root-path" in self.extra_args:
101             raise ValueError("`--pex-root` flag not allowed. We set its value for you.")
102 
103 
104 class PexPEX(DownloadedExternalTool):
105     """The Pex PEX binary."""
106 
107 
108 @rule
109 async def download_pex_pex(pex_cli: PexCli, platform: Platform) -> PexPEX:
110     pex_pex = await Get(DownloadedExternalTool, ExternalToolRequest, pex_cli.get_request(platform))
111     return PexPEX(digest=pex_pex.digest, exe=pex_pex.exe)
112 
113 
114 @rule
115 async def setup_pex_cli_process(
116     request: PexCliProcess,
117     pex_pex: PexPEX,
118     pex_env: PexEnvironment,
119     bootstrap_python: PythonBuildStandaloneBinary,
120     python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,
121     global_options: GlobalOptions,
122     pex_subsystem: PexSubsystem,
123     python_setup: PythonSetup,
124 ) -> Process:
125     tmpdir = ".tmp"
126     gets: List[Get] = [Get(Digest, CreateDigest([Directory(tmpdir)]))]
127 
128     cert_args = []
129     if global_options.ca_certs_path:
130         ca_certs_fc = ca_certs_path_to_file_content(global_options.ca_certs_path)
131         gets.append(Get(Digest, CreateDigest((ca_certs_fc,))))
132         cert_args = ["--cert", ca_certs_fc.path]
133 
134     digests_to_merge = [pex_pex.digest]
135     digests_to_merge.extend(await MultiGet(gets))
136     if request.additional_input_digest:
137         digests_to_merge.append(request.additional_input_digest)
138     input_digest = await Get(Digest, MergeDigests(digests_to_merge))
139 
140     global_args = [
141         # Ensure Pex and its subprocesses create temporary files in the process execution
142         # sandbox. It may make sense to do this generally for Processes, but in the short term we
143         # have known use cases where /tmp is too small to hold large wheel downloads Pex is asked to
144         # perform. Making the TMPDIR local to the sandbox allows control via
145         # --local-execution-root-dir for the local case and should work well with remote cases where
146         # a remoting implementation has to allow for processes producing large binaries in a
147         # sandbox to support reasonable workloads. Communicating TMPDIR via --tmpdir instead of via
148         # environment variable allows Pex to absolutize the path ensuring subprocesses that change
149         # CWD can find the TMPDIR.
150         "--tmpdir",
151         tmpdir,
152     ]
153 
154     if request.concurrency_available > 0:
155         global_args.extend(["--jobs", "{pants_concurrency}"])
156 
157     verbosity_args = [f"-{'v' * pex_subsystem.verbosity}"] if pex_subsystem.verbosity > 0 else []
158 
159     # NB: We should always pass `--python-path`, as that tells Pex where to look for interpreters
160     # when `--python` isn't an absolute path.
161     resolve_args = [
162         *cert_args,
163         "--python-path",
164         os.pathsep.join(pex_env.interpreter_search_paths),
165     ]
166     # All old-style pex runs take the --pip-version flag, but only certain subcommands of the
167     # `pex3` console script do. So if invoked with a subcommand, the caller must selectively
168     # set --pip-version only on subcommands that take it.
169     pip_version_args = [] if request.subcommand else ["--pip-version", python_setup.pip_version]
170     args = [
171         *request.subcommand,
172         *global_args,
173         *verbosity_args,
174         *pip_version_args,
175         *resolve_args,
176         # NB: This comes at the end because it may use `--` passthrough args, # which must come at
177         # the end.
178         *request.extra_args,
179     ]
180 
181     complete_pex_env = pex_env.in_sandbox(working_directory=None)
182     normalized_argv = complete_pex_env.create_argv(pex_pex.exe, *args)
183     env = {
184         **complete_pex_env.environment_dict(python=bootstrap_python),
185         **python_native_code.subprocess_env_vars,
186         **(request.extra_env or {}),
187         # If a subcommand is used, we need to use the `pex3` console script.
188         **({"PEX_SCRIPT": "pex3"} if request.subcommand else {}),
189     }
190 
191     return Process(
192         normalized_argv,
193         description=request.description,
194         input_digest=input_digest,
195         env=env,
196         output_files=request.output_files,
197         output_directories=request.output_directories,
198         append_only_caches=complete_pex_env.append_only_caches,
199         level=request.level,
200         concurrency_available=request.concurrency_available,
201         cache_scope=request.cache_scope,
202     )
203 
204 
205 def rules():
206     return [
207         *collect_rules(),
208         *external_tool.rules(),
209         *pex_environment.rules(),
210         *adhoc_binaries.rules(),
211     ]
212
```
--- 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/src/python/pants/backend/python/util_rules/pex_cli.py b/src/python/pants/backend/python/util_rules/pex_cli.py
--- a/src/python/pants/backend/python/util_rules/pex_cli.py
+++ b/src/python/pants/backend/python/util_rules/pex_cli.py
@@ -35,7 +35,7 @@
     name = "pex"
     help = "The PEX (Python EXecutable) tool (https://github.com/pantsbuild/pex)."
 
-    default_version = "v2.1.159"
+    default_version = "v2.1.162"
     default_url_template = "https://github.com/pantsbuild/pex/releases/download/{version}/pex"
     version_constraints = ">=2.1.135,<3.0"
 
@@ -46,8 +46,8 @@
                 (
                     cls.default_version,
                     plat,
-                    "83c3090938b4d276703864c34ba50bcb3616db0663c54b56dd0521a668d9555f",
-                    "3671772",
+                    "95babb1aa147e2803b7744ba0c025aede8e5fc00322ed535675a51563672486b",
+                    "3676166",
                 )
             )
             for plat in ["macos_arm64", "macos_x86_64", "linux_x86_64", "linux_arm64"]

{"golden_diff": "diff --git a/src/python/pants/backend/python/util_rules/pex_cli.py b/src/python/pants/backend/python/util_rules/pex_cli.py\n--- a/src/python/pants/backend/python/util_rules/pex_cli.py\n+++ b/src/python/pants/backend/python/util_rules/pex_cli.py\n@@ -35,7 +35,7 @@\n     name = \"pex\"\n     help = \"The PEX (Python EXecutable) tool (https://github.com/pantsbuild/pex).\"\n \n-    default_version = \"v2.1.159\"\n+    default_version = \"v2.1.162\"\n     default_url_template = \"https://github.com/pantsbuild/pex/releases/download/{version}/pex\"\n     version_constraints = \">=2.1.135,<3.0\"\n \n@@ -46,8 +46,8 @@\n                 (\n                     cls.default_version,\n                     plat,\n-                    \"83c3090938b4d276703864c34ba50bcb3616db0663c54b56dd0521a668d9555f\",\n-                    \"3671772\",\n+                    \"95babb1aa147e2803b7744ba0c025aede8e5fc00322ed535675a51563672486b\",\n+                    \"3676166\",\n                 )\n             )\n             for plat in [\"macos_arm64\", \"macos_x86_64\", \"linux_x86_64\", \"linux_arm64\"]\n", "issue": "Pants 2.19 fails to download WHLs with URL-escaped chars\n**Describe the bug**\r\nPants 2.19 fails to download WHL files hosted on URLs that contain URL-escaped characters. This applies to both URLs already present in the lockfile and to URLs retrieved during `generate-lockfiles`.\r\n\r\nFor example, instead of:\r\nhttps://download.pytorch.org/whl/cpu/torch-2.0.1%2Bcpu-cp311-cp311-linux_x86_64.whl\r\nit attempts (and fails) to download:\r\nhttps://download.pytorch.org/whl/cpu/torch-2.0.1+cpu-cp311-cp311-linux_x86_64.whl\r\n\r\nSeems to be related to the `v2.1.148` PEX version, as downgrading to `v2.1.137` resolves the issue\r\n\r\n**Pants version**\r\n2.19.0\r\n\r\n**OS**\r\nBoth MacOS and Linux\r\n\r\n**Additional info**\r\nMinimalist example to reproduce:\r\n\r\n```BUILD\r\npython_requirement(\r\n    requirements=[\"torch==2.0.1+cpu\"],\r\n)\r\n```\r\n\r\n```pants.toml\r\n[GLOBAL]\r\npants_version = \"2.19.0\"\r\n\r\nbackend_packages = [\r\n  \"pants.backend.python\",\r\n]\r\n\r\n[python]\r\ninterpreter_constraints = [\"==3.11.*\"]\r\nenable_resolves = true\r\n\r\n[python-repos]\r\nfind_links = [\r\n    \"https://download.pytorch.org/whl/torch_stable.html\",\r\n]\r\n\r\n# Workaround: Downgrade to a previous PEX version\r\n#[pex-cli]\r\n#known_versions = [\r\n#  \"v2.1.137|macos_arm64|faad51a6a108fba9d40b2a10e82a2646fccbaf8c3d9be47818f4bffae02d94b8|4098329\",\r\n#  \"v2.1.137|macos_x86_64|faad51a6a108fba9d40b2a10e82a2646fccbaf8c3d9be47818f4bffae02d94b8|4098329\",\r\n#  \"v2.1.137|linux_x86_64|faad51a6a108fba9d40b2a10e82a2646fccbaf8c3d9be47818f4bffae02d94b8|4098329\",\r\n#  \"v2.1.137|linux_arm64|faad51a6a108fba9d40b2a10e82a2646fccbaf8c3d9be47818f4bffae02d94b8|4098329\"\r\n#]\r\n#version = \"v2.1.137\"\r\n\r\n```\n", "before_files": [{"content": "# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).\n# Licensed under the Apache License, Version 2.0 (see LICENSE).\n\nfrom __future__ import annotations\n\nimport dataclasses\nimport os.path\nfrom dataclasses import dataclass\nfrom typing import Iterable, List, Mapping, Optional, Tuple\n\nfrom pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem\nfrom pants.backend.python.subsystems.setup import PythonSetup\nfrom pants.backend.python.util_rules import pex_environment\nfrom pants.backend.python.util_rules.pex_environment import PexEnvironment, PexSubsystem\nfrom pants.core.util_rules import adhoc_binaries, external_tool\nfrom pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary\nfrom pants.core.util_rules.external_tool import (\n    DownloadedExternalTool,\n    ExternalToolRequest,\n    TemplatedExternalTool,\n)\nfrom pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests\nfrom pants.engine.internals.selectors import MultiGet\nfrom pants.engine.platform import Platform\nfrom pants.engine.process import Process, ProcessCacheScope\nfrom pants.engine.rules import Get, collect_rules, rule\nfrom pants.option.global_options import GlobalOptions, ca_certs_path_to_file_content\nfrom pants.util.frozendict import FrozenDict\nfrom pants.util.logging import LogLevel\nfrom pants.util.meta import classproperty\n\n\nclass PexCli(TemplatedExternalTool):\n    options_scope = \"pex-cli\"\n    name = \"pex\"\n    help = \"The PEX (Python EXecutable) tool (https://github.com/pantsbuild/pex).\"\n\n    default_version = \"v2.1.159\"\n    default_url_template = \"https://github.com/pantsbuild/pex/releases/download/{version}/pex\"\n    version_constraints = \">=2.1.135,<3.0\"\n\n    @classproperty\n    def default_known_versions(cls):\n        return [\n            \"|\".join(\n                (\n                    cls.default_version,\n                    plat,\n                    \"83c3090938b4d276703864c34ba50bcb3616db0663c54b56dd0521a668d9555f\",\n                    \"3671772\",\n                )\n            )\n            for plat in [\"macos_arm64\", \"macos_x86_64\", \"linux_x86_64\", \"linux_arm64\"]\n        ]\n\n\n@dataclass(frozen=True)\nclass PexCliProcess:\n    subcommand: tuple[str, ...]\n    extra_args: tuple[str, ...]\n    description: str = dataclasses.field(compare=False)\n    additional_input_digest: Optional[Digest]\n    extra_env: Optional[FrozenDict[str, str]]\n    output_files: Optional[Tuple[str, ...]]\n    output_directories: Optional[Tuple[str, ...]]\n    level: LogLevel\n    concurrency_available: int\n    cache_scope: ProcessCacheScope\n\n    def __init__(\n        self,\n        *,\n        subcommand: Iterable[str],\n        extra_args: Iterable[str],\n        description: str,\n        additional_input_digest: Optional[Digest] = None,\n        extra_env: Optional[Mapping[str, str]] = None,\n        output_files: Optional[Iterable[str]] = None,\n        output_directories: Optional[Iterable[str]] = None,\n        level: LogLevel = LogLevel.INFO,\n        concurrency_available: int = 0,\n        cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,\n    ) -> None:\n        object.__setattr__(self, \"subcommand\", tuple(subcommand))\n        object.__setattr__(self, \"extra_args\", tuple(extra_args))\n        object.__setattr__(self, \"description\", description)\n        object.__setattr__(self, \"additional_input_digest\", additional_input_digest)\n        object.__setattr__(self, \"extra_env\", FrozenDict(extra_env) if extra_env else None)\n        object.__setattr__(self, \"output_files\", tuple(output_files) if output_files else None)\n        object.__setattr__(\n            self, \"output_directories\", tuple(output_directories) if output_directories else None\n        )\n        object.__setattr__(self, \"level\", level)\n        object.__setattr__(self, \"concurrency_available\", concurrency_available)\n        object.__setattr__(self, \"cache_scope\", cache_scope)\n\n        self.__post_init__()\n\n    def __post_init__(self) -> None:\n        if \"--pex-root-path\" in self.extra_args:\n            raise ValueError(\"`--pex-root` flag not allowed. We set its value for you.\")\n\n\nclass PexPEX(DownloadedExternalTool):\n    \"\"\"The Pex PEX binary.\"\"\"\n\n\n@rule\nasync def download_pex_pex(pex_cli: PexCli, platform: Platform) -> PexPEX:\n    pex_pex = await Get(DownloadedExternalTool, ExternalToolRequest, pex_cli.get_request(platform))\n    return PexPEX(digest=pex_pex.digest, exe=pex_pex.exe)\n\n\n@rule\nasync def setup_pex_cli_process(\n    request: PexCliProcess,\n    pex_pex: PexPEX,\n    pex_env: PexEnvironment,\n    bootstrap_python: PythonBuildStandaloneBinary,\n    python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,\n    global_options: GlobalOptions,\n    pex_subsystem: PexSubsystem,\n    python_setup: PythonSetup,\n) -> Process:\n    tmpdir = \".tmp\"\n    gets: List[Get] = [Get(Digest, CreateDigest([Directory(tmpdir)]))]\n\n    cert_args = []\n    if global_options.ca_certs_path:\n        ca_certs_fc = ca_certs_path_to_file_content(global_options.ca_certs_path)\n        gets.append(Get(Digest, CreateDigest((ca_certs_fc,))))\n        cert_args = [\"--cert\", ca_certs_fc.path]\n\n    digests_to_merge = [pex_pex.digest]\n    digests_to_merge.extend(await MultiGet(gets))\n    if request.additional_input_digest:\n        digests_to_merge.append(request.additional_input_digest)\n    input_digest = await Get(Digest, MergeDigests(digests_to_merge))\n\n    global_args = [\n        # Ensure Pex and its subprocesses create temporary files in the process execution\n        # sandbox. It may make sense to do this generally for Processes, but in the short term we\n        # have known use cases where /tmp is too small to hold large wheel downloads Pex is asked to\n        # perform. Making the TMPDIR local to the sandbox allows control via\n        # --local-execution-root-dir for the local case and should work well with remote cases where\n        # a remoting implementation has to allow for processes producing large binaries in a\n        # sandbox to support reasonable workloads. Communicating TMPDIR via --tmpdir instead of via\n        # environment variable allows Pex to absolutize the path ensuring subprocesses that change\n        # CWD can find the TMPDIR.\n        \"--tmpdir\",\n        tmpdir,\n    ]\n\n    if request.concurrency_available > 0:\n        global_args.extend([\"--jobs\", \"{pants_concurrency}\"])\n\n    verbosity_args = [f\"-{'v' * pex_subsystem.verbosity}\"] if pex_subsystem.verbosity > 0 else []\n\n    # NB: We should always pass `--python-path`, as that tells Pex where to look for interpreters\n    # when `--python` isn't an absolute path.\n    resolve_args = [\n        *cert_args,\n        \"--python-path\",\n        os.pathsep.join(pex_env.interpreter_search_paths),\n    ]\n    # All old-style pex runs take the --pip-version flag, but only certain subcommands of the\n    # `pex3` console script do. So if invoked with a subcommand, the caller must selectively\n    # set --pip-version only on subcommands that take it.\n    pip_version_args = [] if request.subcommand else [\"--pip-version\", python_setup.pip_version]\n    args = [\n        *request.subcommand,\n        *global_args,\n        *verbosity_args,\n        *pip_version_args,\n        *resolve_args,\n        # NB: This comes at the end because it may use `--` passthrough args, # which must come at\n        # the end.\n        *request.extra_args,\n    ]\n\n    complete_pex_env = pex_env.in_sandbox(working_directory=None)\n    normalized_argv = complete_pex_env.create_argv(pex_pex.exe, *args)\n    env = {\n        **complete_pex_env.environment_dict(python=bootstrap_python),\n        **python_native_code.subprocess_env_vars,\n        **(request.extra_env or {}),\n        # If a subcommand is used, we need to use the `pex3` console script.\n        **({\"PEX_SCRIPT\": \"pex3\"} if request.subcommand else {}),\n    }\n\n    return Process(\n        normalized_argv,\n        description=request.description,\n        input_digest=input_digest,\n        env=env,\n        output_files=request.output_files,\n        output_directories=request.output_directories,\n        append_only_caches=complete_pex_env.append_only_caches,\n        level=request.level,\n        concurrency_available=request.concurrency_available,\n        cache_scope=request.cache_scope,\n    )\n\n\ndef rules():\n    return [\n        *collect_rules(),\n        *external_tool.rules(),\n        *pex_environment.rules(),\n        *adhoc_binaries.rules(),\n    ]\n", "path": "src/python/pants/backend/python/util_rules/pex_cli.py"}], "after_files": [{"content": "# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).\n# Licensed under the Apache License, Version 2.0 (see LICENSE).\n\nfrom __future__ import annotations\n\nimport dataclasses\nimport os.path\nfrom dataclasses import dataclass\nfrom typing import Iterable, List, Mapping, Optional, Tuple\n\nfrom pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem\nfrom pants.backend.python.subsystems.setup import PythonSetup\nfrom pants.backend.python.util_rules import pex_environment\nfrom pants.backend.python.util_rules.pex_environment import PexEnvironment, PexSubsystem\nfrom pants.core.util_rules import adhoc_binaries, external_tool\nfrom pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary\nfrom pants.core.util_rules.external_tool import (\n    DownloadedExternalTool,\n    ExternalToolRequest,\n    TemplatedExternalTool,\n)\nfrom pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests\nfrom pants.engine.internals.selectors import MultiGet\nfrom pants.engine.platform import Platform\nfrom pants.engine.process import Process, ProcessCacheScope\nfrom pants.engine.rules import Get, collect_rules, rule\nfrom pants.option.global_options import GlobalOptions, ca_certs_path_to_file_content\nfrom pants.util.frozendict import FrozenDict\nfrom pants.util.logging import LogLevel\nfrom pants.util.meta import classproperty\n\n\nclass PexCli(TemplatedExternalTool):\n    options_scope = \"pex-cli\"\n    name = \"pex\"\n    help = \"The PEX (Python EXecutable) tool (https://github.com/pantsbuild/pex).\"\n\n    default_version = \"v2.1.162\"\n    default_url_template = \"https://github.com/pantsbuild/pex/releases/download/{version}/pex\"\n    version_constraints = \">=2.1.135,<3.0\"\n\n    @classproperty\n    def default_known_versions(cls):\n        return [\n            \"|\".join(\n                (\n                    cls.default_version,\n                    plat,\n                    \"95babb1aa147e2803b7744ba0c025aede8e5fc00322ed535675a51563672486b\",\n                    \"3676166\",\n                )\n            )\n            for plat in [\"macos_arm64\", \"macos_x86_64\", \"linux_x86_64\", \"linux_arm64\"]\n        ]\n\n\n@dataclass(frozen=True)\nclass PexCliProcess:\n    subcommand: tuple[str, ...]\n    extra_args: tuple[str, ...]\n    description: str = dataclasses.field(compare=False)\n    additional_input_digest: Optional[Digest]\n    extra_env: Optional[FrozenDict[str, str]]\n    output_files: Optional[Tuple[str, ...]]\n    output_directories: Optional[Tuple[str, ...]]\n    level: LogLevel\n    concurrency_available: int\n    cache_scope: ProcessCacheScope\n\n    def __init__(\n        self,\n        *,\n        subcommand: Iterable[str],\n        extra_args: Iterable[str],\n        description: str,\n        additional_input_digest: Optional[Digest] = None,\n        extra_env: Optional[Mapping[str, str]] = None,\n        output_files: Optional[Iterable[str]] = None,\n        output_directories: Optional[Iterable[str]] = None,\n        level: LogLevel = LogLevel.INFO,\n        concurrency_available: int = 0,\n        cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,\n    ) -> None:\n        object.__setattr__(self, \"subcommand\", tuple(subcommand))\n        object.__setattr__(self, \"extra_args\", tuple(extra_args))\n        object.__setattr__(self, \"description\", description)\n        object.__setattr__(self, \"additional_input_digest\", additional_input_digest)\n        object.__setattr__(self, \"extra_env\", FrozenDict(extra_env) if extra_env else None)\n        object.__setattr__(self, \"output_files\", tuple(output_files) if output_files else None)\n        object.__setattr__(\n            self, \"output_directories\", tuple(output_directories) if output_directories else None\n        )\n        object.__setattr__(self, \"level\", level)\n        object.__setattr__(self, \"concurrency_available\", concurrency_available)\n        object.__setattr__(self, \"cache_scope\", cache_scope)\n\n        self.__post_init__()\n\n    def __post_init__(self) -> None:\n        if \"--pex-root-path\" in self.extra_args:\n            raise ValueError(\"`--pex-root` flag not allowed. We set its value for you.\")\n\n\nclass PexPEX(DownloadedExternalTool):\n    \"\"\"The Pex PEX binary.\"\"\"\n\n\n@rule\nasync def download_pex_pex(pex_cli: PexCli, platform: Platform) -> PexPEX:\n    pex_pex = await Get(DownloadedExternalTool, ExternalToolRequest, pex_cli.get_request(platform))\n    return PexPEX(digest=pex_pex.digest, exe=pex_pex.exe)\n\n\n@rule\nasync def setup_pex_cli_process(\n    request: PexCliProcess,\n    pex_pex: PexPEX,\n    pex_env: PexEnvironment,\n    bootstrap_python: PythonBuildStandaloneBinary,\n    python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,\n    global_options: GlobalOptions,\n    pex_subsystem: PexSubsystem,\n    python_setup: PythonSetup,\n) -> Process:\n    tmpdir = \".tmp\"\n    gets: List[Get] = [Get(Digest, CreateDigest([Directory(tmpdir)]))]\n\n    cert_args = []\n    if global_options.ca_certs_path:\n        ca_certs_fc = ca_certs_path_to_file_content(global_options.ca_certs_path)\n        gets.append(Get(Digest, CreateDigest((ca_certs_fc,))))\n        cert_args = [\"--cert\", ca_certs_fc.path]\n\n    digests_to_merge = [pex_pex.digest]\n    digests_to_merge.extend(await MultiGet(gets))\n    if request.additional_input_digest:\n        digests_to_merge.append(request.additional_input_digest)\n    input_digest = await Get(Digest, MergeDigests(digests_to_merge))\n\n    global_args = [\n        # Ensure Pex and its subprocesses create temporary files in the process execution\n        # sandbox. It may make sense to do this generally for Processes, but in the short term we\n        # have known use cases where /tmp is too small to hold large wheel downloads Pex is asked to\n        # perform. Making the TMPDIR local to the sandbox allows control via\n        # --local-execution-root-dir for the local case and should work well with remote cases where\n        # a remoting implementation has to allow for processes producing large binaries in a\n        # sandbox to support reasonable workloads. Communicating TMPDIR via --tmpdir instead of via\n        # environment variable allows Pex to absolutize the path ensuring subprocesses that change\n        # CWD can find the TMPDIR.\n        \"--tmpdir\",\n        tmpdir,\n    ]\n\n    if request.concurrency_available > 0:\n        global_args.extend([\"--jobs\", \"{pants_concurrency}\"])\n\n    verbosity_args = [f\"-{'v' * pex_subsystem.verbosity}\"] if pex_subsystem.verbosity > 0 else []\n\n    # NB: We should always pass `--python-path`, as that tells Pex where to look for interpreters\n    # when `--python` isn't an absolute path.\n    resolve_args = [\n        *cert_args,\n        \"--python-path\",\n        os.pathsep.join(pex_env.interpreter_search_paths),\n    ]\n    # All old-style pex runs take the --pip-version flag, but only certain subcommands of the\n    # `pex3` console script do. So if invoked with a subcommand, the caller must selectively\n    # set --pip-version only on subcommands that take it.\n    pip_version_args = [] if request.subcommand else [\"--pip-version\", python_setup.pip_version]\n    args = [\n        *request.subcommand,\n        *global_args,\n        *verbosity_args,\n        *pip_version_args,\n        *resolve_args,\n        # NB: This comes at the end because it may use `--` passthrough args, # which must come at\n        # the end.\n        *request.extra_args,\n    ]\n\n    complete_pex_env = pex_env.in_sandbox(working_directory=None)\n    normalized_argv = complete_pex_env.create_argv(pex_pex.exe, *args)\n    env = {\n        **complete_pex_env.environment_dict(python=bootstrap_python),\n        **python_native_code.subprocess_env_vars,\n        **(request.extra_env or {}),\n        # If a subcommand is used, we need to use the `pex3` console script.\n        **({\"PEX_SCRIPT\": \"pex3\"} if request.subcommand else {}),\n    }\n\n    return Process(\n        normalized_argv,\n        description=request.description,\n        input_digest=input_digest,\n        env=env,\n        output_files=request.output_files,\n        output_directories=request.output_directories,\n        append_only_caches=complete_pex_env.append_only_caches,\n        level=request.level,\n        concurrency_available=request.concurrency_available,\n        cache_scope=request.cache_scope,\n    )\n\n\ndef rules():\n    return [\n        *collect_rules(),\n        *external_tool.rules(),\n        *pex_environment.rules(),\n        *adhoc_binaries.rules(),\n    ]\n", "path": "src/python/pants/backend/python/util_rules/pex_cli.py"}]}