20390128985

Committed 20 Dec 2025 05:59AM UTC coverage: 80.275% (-0.02%) from 80.296%

Build # 20390128985

Build Type

Pull #21991

github

Committed by

web-flow

Commit Message

Merge 9294b7195 into 8e0c2b8b0

Pull Request Pull Request #21991: [WIP] use `uv` for resolving Pants plugins

Run Details

69 of 109 new or added lines in 4 files covered. (63.3%)

3 existing lines in 2 files now uncovered.

78553 of 97855 relevant lines covered (80.27%)

3.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

66.17

/src/python/pants/init/plugin_resolver.py

# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import importlib.metadata
import logging
import site
import sys
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from io import StringIO
from typing import cast

from packaging.requirements import Requirement

from pants.backend.python.subsystems.repos import PythonRepos
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.pex import (
    PexRequest,
    VenvPexProcess,
    create_venv_pex,
)
from pants.backend.python.util_rules.pex_environment import PythonExecutable
from pants.backend.python.util_rules.pex_requirements import PexRequirements
from pants.core.environments.rules import determine_bootstrap_environment
from pants.core.subsystems.uv import UvTool
from pants.core.subsystems.uv import rules as uv_rules
from pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary
from pants.core.util_rules.adhoc_binaries import rules as adhoc_binaries_rules
from pants.core.util_rules.env_vars import environment_vars_subset
from pants.engine.collection import DeduplicatedCollection
from pants.engine.env_vars import CompleteEnvironmentVars, EnvironmentVarsRequest
from pants.engine.environment import EnvironmentName
from pants.engine.fs import CreateDigest, Digest, FileContent, MergeDigests
from pants.engine.internals.selectors import Params
from pants.engine.internals.session import SessionValues
from pants.engine.intrinsics import create_digest, execute_process, merge_digests
from pants.engine.platform import Platform
from pants.engine.process import (
    Process,
    ProcessCacheScope,
    ProcessExecutionEnvironment,
    execute_process_or_raise,
)
from pants.engine.rules import QueryRule, collect_rules, implicitly, rule
from pants.init.bootstrap_scheduler import BootstrapScheduler
from pants.init.import_util import find_matching_distributions
from pants.option.global_options import GlobalOptions
from pants.option.options_bootstrapper import OptionsBootstrapper
from pants.util.logging import LogLevel

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class PluginsRequest:
    # Interpreter constraints to resolve for, or None to resolve for the interpreter that Pants is
    # running under.
    interpreter_constraints: InterpreterConstraints | None
    # Requirement constraints to resolve with. If plugins will be loaded into the global working_set
    # (i.e., onto the `sys.path`), then these should be the current contents of the working_set.
    constraints: tuple[Requirement, ...]
    # Backend requirements to resolve
    requirements: tuple[str, ...]


class ResolvedPluginDistributions(DeduplicatedCollection[str]):
    sort_input = True


class ResolvedPluginDistributionsPEX(ResolvedPluginDistributions):
    pass


class ResolvedPluginDistributionsUV(ResolvedPluginDistributions):
    pass


@rule
async def resolve_plugins_via_pex(
    request: PluginsRequest, global_options: GlobalOptions
) -> ResolvedPluginDistributionsPEX:
    """This rule resolves plugins using a VenvPex, and exposes the absolute paths of their dists.

    NB: This relies on the fact that PEX constructs venvs in a stable location (within the
    `named_caches` directory), but consequently needs to disable the process cache: see the
    ProcessCacheScope reference in the body.
    """
    logger.info("resolve_plugins_via_pex")

    req_strings = sorted(global_options.plugins + request.requirements)

    requirements = PexRequirements(
        req_strings_or_addrs=req_strings,
        constraints_strings=(str(constraint) for constraint in request.constraints),
        description_of_origin="configured Pants plugins",
    )
    if not requirements:
        return ResolvedPluginDistributionsPEX()

    python: PythonExecutable | None = None
    if not request.interpreter_constraints:
        python = PythonExecutable.fingerprinted(
            sys.executable, ".".join(map(str, sys.version_info[:3])).encode("utf8")
        )

    plugins_pex = await create_venv_pex(
        **implicitly(
            PexRequest(
                output_filename="pants_plugins.pex",
                internal_only=True,
                python=python,
                requirements=requirements,
                interpreter_constraints=request.interpreter_constraints or InterpreterConstraints(),
                description=f"Resolving plugins: {', '.join(req_strings)}",
            )
        )
    )

    # NB: We run this Process per-restart because it (intentionally) leaks named cache
    # paths in a way that invalidates the Process-cache. See the method doc.
    cache_scope = (
        ProcessCacheScope.PER_SESSION
        if global_options.plugins_force_resolve
        else ProcessCacheScope.PER_RESTART_SUCCESSFUL
    )

    plugins_process_result = await execute_process_or_raise(
        **implicitly(
            VenvPexProcess(
                plugins_pex,
                argv=("-c", "import os, site; print(os.linesep.join(site.getsitepackages()))"),
                description="Extracting plugin locations",
                level=LogLevel.DEBUG,
                cache_scope=cache_scope,
            )
        )
    )
    return ResolvedPluginDistributionsPEX(
        plugins_process_result.stdout.decode().strip().split("\n")
    )


@dataclass(frozen=True)
class _UvPluginResolveScript:
    digest: Digest
    path: str


# Script which invokes `uv` to resolve plugin distributions. It builds a mini-uv project in a subdirectory
# of the repository and uses `uv` to manage the venv in that project.
_UV_PLUGIN_RESOLVE_SCRIPT = r"""\
import os
from pathlib import Path
import shutil
from subprocess import run
import sys


def pyproject_changed(current_path: Path, previous_path: Path) -> bool:
    with open(current_path, "rb") as f:
        current_pyproject = f.read()

    with open(previous_path, "rb") as f:
        previous_pyproject = f.read()

    return current_pyproject != previous_pyproject


inputs_dir = Path(sys.argv[1])
uv_path = inputs_dir / sys.argv[2]
pyproject_path = inputs_dir / sys.argv[3]

plugins_path = Path(".pants.d/plugins")
plugins_path.mkdir(parents=True, exist_ok=True)
os.chdir(plugins_path)

shutil.copy(pyproject_path, ".")

run([uv_path, "sync", f"--python={sys.executable}"], check=True)
run(["./.venv/bin/python", "-c", "import os, site; print(os.linesep.join(site.getsitepackages()))"], check=True)
"""


@rule
async def _setup_uv_plugin_resolve_script() -> _UvPluginResolveScript:
    digest = await create_digest(
        CreateDigest(
            [FileContent(content=_UV_PLUGIN_RESOLVE_SCRIPT.encode(), path="uv_plugin_resolve.py")]
        ),
    )
    return _UvPluginResolveScript(digest=digest, path="uv_plugin_resolve.py")


_PYPROJECT_TEMPLATE = """\
[project]
name = "pants-plugins"
version = "0.0.1"
description = "Plugins for your Pants"
requires-python = "==3.11.*"
dependencies = [{requirements_formatted}]
[tool.uv]
package = false
fork-strategy = "requires-python"
environments = ["sys_platform == '{platform}'"]
constraint-dependencies = [{constraints_formatted}]
find-links = [{find_links_formatted}]
{indexes_formatted}
[tool.__pants_internal__]
version = {version}
"""


def _generate_pyproject_toml(
    requirements: Iterable[str],
    constraints: Iterable[str],
    python_indexes: Sequence[str],
    python_find_links: Iterable[str],
) -> str:
    requirements_formatted = ", ".join([f'"{x}"' for x in requirements])
    constraints_formatted = ", ".join([f'"{x}"' for x in constraints])
    find_links_formatted = ", ".join([f'"{x}"' for x in python_find_links])

    indexes_formatted = StringIO()
    if python_indexes:
        for python_index in python_indexes:
            indexes_formatted.write(f"""[[tool.uv.index]]\nurl = "{python_index}"\n""")
        indexes_formatted.write("default = true")
    else:
        indexes_formatted.write("no-index = true")

    return _PYPROJECT_TEMPLATE.format(
        constraints_formatted=constraints_formatted,
        find_links_formatted=find_links_formatted,
        indexes_formatted=indexes_formatted.getvalue(),
        platform=sys.platform,
        requirements_formatted=requirements_formatted,
        version=0,
    )


@rule
async def resolve_plugins_via_uv(
    request: PluginsRequest,
    global_options: GlobalOptions,
    python_repos: PythonRepos,
    uv_tool: UvTool,
    uv_plugin_resolve_script: _UvPluginResolveScript,
    platform: Platform,
    python_binary: PythonBuildStandaloneBinary,
) -> ResolvedPluginDistributionsUV:
    req_strings = sorted(global_options.plugins + request.requirements)
    if not req_strings:
        return ResolvedPluginDistributionsUV()

    pyproject_content = _generate_pyproject_toml(
        requirements=req_strings,
        constraints=(str(c) for c in request.constraints),
        python_indexes=python_repos.indexes,
        python_find_links=python_repos.find_links,
    )

    data_digest = await create_digest(
        CreateDigest(
            [
                FileContent(content=pyproject_content.encode(), path="pyproject.toml"),
            ]
        ),
        **implicitly(),
    )

    cache_scope = (
        ProcessCacheScope.PER_SESSION
        if global_options.plugins_force_resolve
        else ProcessCacheScope.PER_RESTART_SUCCESSFUL
    )

    input_digest = await merge_digests(
        MergeDigests([uv_plugin_resolve_script.digest, uv_tool.digest, data_digest]),
        **implicitly(),
    )

    env = await environment_vars_subset(
        EnvironmentVarsRequest(["PATH", "HOME"], allowed=["PATH", "HOME"]), **implicitly()
    )

    process = Process(
        argv=(
            python_binary.path,
            f"{{chroot}}/{uv_plugin_resolve_script.path}",
            "{chroot}",
            f"{uv_tool.exe}",
            "pyproject.toml",
        ),
        env=env,
        input_digest=input_digest,
        append_only_caches=python_binary.APPEND_ONLY_CACHES,
        description=f"Resolving plugins: {', '.join(req_strings)}",
        cache_scope=cache_scope,
    )

    workspace_process_execution_environment = ProcessExecutionEnvironment(
        environment_name=None,
        platform=platform.value,
        docker_image=None,
        remote_execution=False,
        remote_execution_extra_platform_properties=(),
        execute_in_workspace=True,
        keep_sandboxes="never",
    )

    result = await execute_process(process, workspace_process_execution_environment)
    print(f"uv result stdout:\n{result.stdout.decode()}")
    print(f"uv result stderr:\n{result.stderr.decode()}")
    if result.exit_code != 0:
        raise ValueError(f"Plugin resolution failed: stderr={result.stderr.decode()}")

    return ResolvedPluginDistributionsUV(result.stdout.decode().strip().split("\n"))


@rule
async def resolve_plugins(
    request: PluginsRequest, global_options: GlobalOptions
) -> ResolvedPluginDistributions:
    if global_options.experimental_use_uv_for_plugin_resolution:
        plugins_from_uv = await resolve_plugins_via_uv(request, **implicitly())
        return ResolvedPluginDistributions(plugins_from_uv)
    else:
        plugins_from_pex = await resolve_plugins_via_pex(request, **implicitly())
        return ResolvedPluginDistributions(plugins_from_pex)


class PluginResolver:
    """Encapsulates the state of plugin loading.

    Plugin loading is inherently stateful, and so the system enviroment on `sys.path` will be
    mutated by each call to `PluginResolver.resolve`.
    """

    def __init__(
        self,
        scheduler: BootstrapScheduler,
        interpreter_constraints: InterpreterConstraints | None = None,
        inherit_existing_constraints: bool = True,
    ) -> None:
        self._scheduler = scheduler
        self._interpreter_constraints = interpreter_constraints
        self._inherit_existing_constraints = inherit_existing_constraints

    def resolve(
        self,
        options_bootstrapper: OptionsBootstrapper,
        env: CompleteEnvironmentVars,
        requirements: Iterable[str] = (),
    ) -> list[str]:
        """Resolves any configured plugins and adds them to the sys.path as a side effect."""

        def to_requirement(d):
            return f"{d.name}=={d.version}"

        distributions: list[importlib.metadata.Distribution] = []
        if self._inherit_existing_constraints:
            distributions = list(find_matching_distributions(None))

        request = PluginsRequest(
            self._interpreter_constraints,
            tuple(to_requirement(dist) for dist in distributions),
            tuple(requirements),
        )

        result = []
        for resolved_plugin_location in self._resolve_plugins(options_bootstrapper, env, request):
            # Activate any .pth files plugin wheels may have.
            orig_sys_path_len = len(sys.path)
            site.addsitedir(resolved_plugin_location)
            if len(sys.path) > orig_sys_path_len:
                result.append(resolved_plugin_location)

        return result

    def _resolve_plugins(
        self,
        options_bootstrapper: OptionsBootstrapper,
        env: CompleteEnvironmentVars,
        request: PluginsRequest,
    ) -> ResolvedPluginDistributions:
        session = self._scheduler.scheduler.new_session(
            "plugin_resolver",
            session_values=SessionValues(
                {
                    OptionsBootstrapper: options_bootstrapper,
                    CompleteEnvironmentVars: env,
                }
            ),
        )
        params = Params(request, determine_bootstrap_environment(session))
        return cast(
            ResolvedPluginDistributions,
            session.product_request(ResolvedPluginDistributions, params)[0],
        )


def rules():
    return [
        QueryRule(ResolvedPluginDistributions, [PluginsRequest, EnvironmentName]),
        *collect_rules(),
        *adhoc_binaries_rules(),
        *uv_rules(),
    ]

1	# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	12✔
5
6	import importlib.metadata	12✔
7	import logging	12✔
8	import site	12✔
9	import sys	12✔
10	from collections.abc import Iterable, Sequence	12✔
11	from dataclasses import dataclass	12✔
12	from io import StringIO	12✔
13	from typing import cast	12✔
14
15	from packaging.requirements import Requirement	12✔
16
17	from pants.backend.python.subsystems.repos import PythonRepos	12✔
18	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	12✔
19	from pants.backend.python.util_rules.pex import (	12✔
20	PexRequest,
21	VenvPexProcess,
22	create_venv_pex,
23	)
24	from pants.backend.python.util_rules.pex_environment import PythonExecutable	12✔
25	from pants.backend.python.util_rules.pex_requirements import PexRequirements	12✔
26	from pants.core.environments.rules import determine_bootstrap_environment	12✔
27	from pants.core.subsystems.uv import UvTool	12✔
28	from pants.core.subsystems.uv import rules as uv_rules	12✔
29	from pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary	12✔
30	from pants.core.util_rules.adhoc_binaries import rules as adhoc_binaries_rules	12✔
31	from pants.core.util_rules.env_vars import environment_vars_subset	12✔
32	from pants.engine.collection import DeduplicatedCollection	12✔
33	from pants.engine.env_vars import CompleteEnvironmentVars, EnvironmentVarsRequest	12✔
34	from pants.engine.environment import EnvironmentName	12✔
35	from pants.engine.fs import CreateDigest, Digest, FileContent, MergeDigests	12✔
36	from pants.engine.internals.selectors import Params	12✔
37	from pants.engine.internals.session import SessionValues	12✔
38	from pants.engine.intrinsics import create_digest, execute_process, merge_digests	12✔
39	from pants.engine.platform import Platform	12✔
40	from pants.engine.process import (	12✔
41	Process,
42	ProcessCacheScope,
43	ProcessExecutionEnvironment,
44	execute_process_or_raise,
45	)
46	from pants.engine.rules import QueryRule, collect_rules, implicitly, rule	12✔
47	from pants.init.bootstrap_scheduler import BootstrapScheduler	12✔
48	from pants.init.import_util import find_matching_distributions	12✔
49	from pants.option.global_options import GlobalOptions	12✔
50	from pants.option.options_bootstrapper import OptionsBootstrapper	12✔
51	from pants.util.logging import LogLevel	12✔
52
53	logger = logging.getLogger(__name__)	12✔
54
55
56	@dataclass(frozen=True)	12✔
57	class PluginsRequest:	12✔
58	# Interpreter constraints to resolve for, or None to resolve for the interpreter that Pants is
59	# running under.
60	interpreter_constraints: InterpreterConstraints \| None	12✔
61	# Requirement constraints to resolve with. If plugins will be loaded into the global working_set
62	# (i.e., onto the `sys.path`), then these should be the current contents of the working_set.
63	constraints: tuple[Requirement, ...]	12✔
64	# Backend requirements to resolve
65	requirements: tuple[str, ...]	12✔
66
67
68	class ResolvedPluginDistributions(DeduplicatedCollection[str]):	12✔
69	sort_input = True	12✔
70
71
72	class ResolvedPluginDistributionsPEX(ResolvedPluginDistributions):	12✔
73	pass	12✔
74
75
76	class ResolvedPluginDistributionsUV(ResolvedPluginDistributions):	12✔
77	pass	12✔
78
79
80	@rule	12✔
81	async def resolve_plugins_via_pex(	12✔
82	request: PluginsRequest, global_options: GlobalOptions
83	) -> ResolvedPluginDistributionsPEX:
84	"""This rule resolves plugins using a VenvPex, and exposes the absolute paths of their dists.
85
86	NB: This relies on the fact that PEX constructs venvs in a stable location (within the
87	`named_caches` directory), but consequently needs to disable the process cache: see the
88	ProcessCacheScope reference in the body.
89	"""
NEW 90	logger.info("resolve_plugins_via_pex")	×
91
UNCOV 92	req_strings = sorted(global_options.plugins + request.requirements)	×
93
94	requirements = PexRequirements(	×
95	req_strings_or_addrs=req_strings,
96	constraints_strings=(str(constraint) for constraint in request.constraints),
97	description_of_origin="configured Pants plugins",
98	)
99	if not requirements:	×
NEW 100	return ResolvedPluginDistributionsPEX()	×
101
102	python: PythonExecutable \| None = None	×
103	if not request.interpreter_constraints:	×
104	python = PythonExecutable.fingerprinted(	×
105	sys.executable, ".".join(map(str, sys.version_info[:3])).encode("utf8")
106	)
107
108	plugins_pex = await create_venv_pex(	×
109	**implicitly(
110	PexRequest(
111	output_filename="pants_plugins.pex",
112	internal_only=True,
113	python=python,
114	requirements=requirements,
115	interpreter_constraints=request.interpreter_constraints or InterpreterConstraints(),
116	description=f"Resolving plugins: {', '.join(req_strings)}",
117	)
118	)
119	)
120
121	# NB: We run this Process per-restart because it (intentionally) leaks named cache
122	# paths in a way that invalidates the Process-cache. See the method doc.
123	cache_scope = (	×
124	ProcessCacheScope.PER_SESSION
125	if global_options.plugins_force_resolve
126	else ProcessCacheScope.PER_RESTART_SUCCESSFUL
127	)
128
129	plugins_process_result = await execute_process_or_raise(	×
130	**implicitly(
131	VenvPexProcess(
132	plugins_pex,
133	argv=("-c", "import os, site; print(os.linesep.join(site.getsitepackages()))"),
134	description="Extracting plugin locations",
135	level=LogLevel.DEBUG,
136	cache_scope=cache_scope,
137	)
138	)
139	)
NEW 140	return ResolvedPluginDistributionsPEX(	×
141	plugins_process_result.stdout.decode().strip().split("\n")
142	)
143
144
145	@dataclass(frozen=True)	12✔
146	class _UvPluginResolveScript:	12✔
147	digest: Digest	12✔
148	path: str	12✔
149
150
151	# Script which invokes `uv` to resolve plugin distributions. It builds a mini-uv project in a subdirectory
152	# of the repository and uses `uv` to manage the venv in that project.
153	_UV_PLUGIN_RESOLVE_SCRIPT = r"""\	12✔
154	import os
155	from pathlib import Path
156	import shutil
157	from subprocess import run
158	import sys
159
160
161	def pyproject_changed(current_path: Path, previous_path: Path) -> bool:
162	with open(current_path, "rb") as f:
163	current_pyproject = f.read()
164
165	with open(previous_path, "rb") as f:
166	previous_pyproject = f.read()
167
168	return current_pyproject != previous_pyproject
169
170
171	inputs_dir = Path(sys.argv[1])
172	uv_path = inputs_dir / sys.argv[2]
173	pyproject_path = inputs_dir / sys.argv[3]
174
175	plugins_path = Path(".pants.d/plugins")
176	plugins_path.mkdir(parents=True, exist_ok=True)
177	os.chdir(plugins_path)
178
179	shutil.copy(pyproject_path, ".")
180
181	run([uv_path, "sync", f"--python={sys.executable}"], check=True)
182	run(["./.venv/bin/python", "-c", "import os, site; print(os.linesep.join(site.getsitepackages()))"], check=True)
183	"""
184
185
186	@rule	12✔
187	async def _setup_uv_plugin_resolve_script() -> _UvPluginResolveScript:	12✔
NEW 188	digest = await create_digest(	×
189	CreateDigest(
190	[FileContent(content=_UV_PLUGIN_RESOLVE_SCRIPT.encode(), path="uv_plugin_resolve.py")]
191	),
192	)
NEW 193	return _UvPluginResolveScript(digest=digest, path="uv_plugin_resolve.py")	×
194
195
196	_PYPROJECT_TEMPLATE = """\	12✔
197	[project]
198	name = "pants-plugins"
199	version = "0.0.1"
200	description = "Plugins for your Pants"
201	requires-python = "==3.11.*"
202	dependencies = [{requirements_formatted}]
203	[tool.uv]
204	package = false
205	fork-strategy = "requires-python"
206	environments = ["sys_platform == '{platform}'"]
207	constraint-dependencies = [{constraints_formatted}]
208	find-links = [{find_links_formatted}]
209	{indexes_formatted}
210	[tool.__pants_internal__]
211	version = {version}
212	"""
213
214
215	def _generate_pyproject_toml(	12✔
216	requirements: Iterable[str],
217	constraints: Iterable[str],
218	python_indexes: Sequence[str],
219	python_find_links: Iterable[str],
220	) -> str:
NEW 221	requirements_formatted = ", ".join([f'"{x}"' for x in requirements])	×
NEW 222	constraints_formatted = ", ".join([f'"{x}"' for x in constraints])	×
NEW 223	find_links_formatted = ", ".join([f'"{x}"' for x in python_find_links])	×
224
NEW 225	indexes_formatted = StringIO()	×
NEW 226	if python_indexes:	×
NEW 227	for python_index in python_indexes:	×
NEW 228	indexes_formatted.write(f"""[[tool.uv.index]]\nurl = "{python_index}"\n""")	×
NEW 229	indexes_formatted.write("default = true")	×
230	else:
NEW 231	indexes_formatted.write("no-index = true")	×
232
NEW 233	return _PYPROJECT_TEMPLATE.format(	×
234	constraints_formatted=constraints_formatted,
235	find_links_formatted=find_links_formatted,
236	indexes_formatted=indexes_formatted.getvalue(),
237	platform=sys.platform,
238	requirements_formatted=requirements_formatted,
239	version=0,
240	)
241
242
243	@rule	12✔
244	async def resolve_plugins_via_uv(	12✔
245	request: PluginsRequest,
246	global_options: GlobalOptions,
247	python_repos: PythonRepos,
248	uv_tool: UvTool,
249	uv_plugin_resolve_script: _UvPluginResolveScript,
250	platform: Platform,
251	python_binary: PythonBuildStandaloneBinary,
252	) -> ResolvedPluginDistributionsUV:
NEW 253	req_strings = sorted(global_options.plugins + request.requirements)	×
NEW 254	if not req_strings:	×
NEW 255	return ResolvedPluginDistributionsUV()	×
256
NEW 257	pyproject_content = _generate_pyproject_toml(	×
258	requirements=req_strings,
259	constraints=(str(c) for c in request.constraints),
260	python_indexes=python_repos.indexes,
261	python_find_links=python_repos.find_links,
262	)
263
NEW 264	data_digest = await create_digest(	×
265	CreateDigest(
266	[
267	FileContent(content=pyproject_content.encode(), path="pyproject.toml"),
268	]
269	),
270	**implicitly(),
271	)
272
NEW 273	cache_scope = (	×
274	ProcessCacheScope.PER_SESSION
275	if global_options.plugins_force_resolve
276	else ProcessCacheScope.PER_RESTART_SUCCESSFUL
277	)
278
NEW 279	input_digest = await merge_digests(	×
280	MergeDigests([uv_plugin_resolve_script.digest, uv_tool.digest, data_digest]),
281	**implicitly(),
282	)
283
NEW 284	env = await environment_vars_subset(	×
285	EnvironmentVarsRequest(["PATH", "HOME"], allowed=["PATH", "HOME"]), **implicitly()
286	)
287
NEW 288	process = Process(	×
289	argv=(
290	python_binary.path,
291	f"{{chroot}}/{uv_plugin_resolve_script.path}",
292	"{chroot}",
293	f"{uv_tool.exe}",
294	"pyproject.toml",
295	),
296	env=env,
297	input_digest=input_digest,
298	append_only_caches=python_binary.APPEND_ONLY_CACHES,
299	description=f"Resolving plugins: {', '.join(req_strings)}",
300	cache_scope=cache_scope,
301	)
302
NEW 303	workspace_process_execution_environment = ProcessExecutionEnvironment(	×
304	environment_name=None,
305	platform=platform.value,
306	docker_image=None,
307	remote_execution=False,
308	remote_execution_extra_platform_properties=(),
309	execute_in_workspace=True,
310	keep_sandboxes="never",
311	)
312
NEW 313	result = await execute_process(process, workspace_process_execution_environment)	×
NEW 314	print(f"uv result stdout:\n{result.stdout.decode()}")	×
NEW 315	print(f"uv result stderr:\n{result.stderr.decode()}")	×
NEW 316	if result.exit_code != 0:	×
NEW 317	raise ValueError(f"Plugin resolution failed: stderr={result.stderr.decode()}")	×
318
NEW 319	return ResolvedPluginDistributionsUV(result.stdout.decode().strip().split("\n"))	×
320
321
322	@rule	12✔
323	async def resolve_plugins(	12✔
324	request: PluginsRequest, global_options: GlobalOptions
325	) -> ResolvedPluginDistributions:
NEW 326	if global_options.experimental_use_uv_for_plugin_resolution:	×
NEW 327	plugins_from_uv = await resolve_plugins_via_uv(request, **implicitly())	×
NEW 328	return ResolvedPluginDistributions(plugins_from_uv)	×
329	else:
NEW 330	plugins_from_pex = await resolve_plugins_via_pex(request, **implicitly())	×
NEW 331	return ResolvedPluginDistributions(plugins_from_pex)	×
332
333
334	class PluginResolver:	12✔
335	"""Encapsulates the state of plugin loading.
336
337	Plugin loading is inherently stateful, and so the system enviroment on `sys.path` will be
338	mutated by each call to `PluginResolver.resolve`.
339	"""
340
341	def __init__(	12✔
342	self,
343	scheduler: BootstrapScheduler,
344	interpreter_constraints: InterpreterConstraints \| None = None,
345	inherit_existing_constraints: bool = True,
346	) -> None:
347	self._scheduler = scheduler	4✔
348	self._interpreter_constraints = interpreter_constraints	4✔
349	self._inherit_existing_constraints = inherit_existing_constraints	4✔
350
351	def resolve(	12✔
352	self,
353	options_bootstrapper: OptionsBootstrapper,
354	env: CompleteEnvironmentVars,
355	requirements: Iterable[str] = (),
356	) -> list[str]:
357	"""Resolves any configured plugins and adds them to the sys.path as a side effect."""
358
359	def to_requirement(d):	4✔
360	return f"{d.name}=={d.version}"	4✔
361
362	distributions: list[importlib.metadata.Distribution] = []	4✔
363	if self._inherit_existing_constraints:	4✔
364	distributions = list(find_matching_distributions(None))	4✔
365
366	request = PluginsRequest(	4✔
367	self._interpreter_constraints,
368	tuple(to_requirement(dist) for dist in distributions),
369	tuple(requirements),
370	)
371
372	result = []	4✔
373	for resolved_plugin_location in self._resolve_plugins(options_bootstrapper, env, request):	4✔
374	# Activate any .pth files plugin wheels may have.
375	orig_sys_path_len = len(sys.path)	1✔
376	site.addsitedir(resolved_plugin_location)	1✔
377	if len(sys.path) > orig_sys_path_len:	1✔
378	result.append(resolved_plugin_location)	1✔
379
380	return result	3✔
381
382	def _resolve_plugins(	12✔
383	self,
384	options_bootstrapper: OptionsBootstrapper,
385	env: CompleteEnvironmentVars,
386	request: PluginsRequest,
387	) -> ResolvedPluginDistributions:
388	session = self._scheduler.scheduler.new_session(	4✔
389	"plugin_resolver",
390	session_values=SessionValues(
391	{
392	OptionsBootstrapper: options_bootstrapper,
393	CompleteEnvironmentVars: env,
394	}
395	),
396	)
397	params = Params(request, determine_bootstrap_environment(session))	4✔
398	return cast(	3✔
399	ResolvedPluginDistributions,
400	session.product_request(ResolvedPluginDistributions, params)[0],
401	)
402
403
404	def rules():	12✔
405	return [	4✔
406	QueryRule(ResolvedPluginDistributions, [PluginsRequest, EnvironmentName]),
407	*collect_rules(),
408	*adhoc_binaries_rules(),
409	*uv_rules(),
410	]

pantsbuild / pants / 20390128985

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous