19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

56.94

/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
from dataclasses import dataclass
from typing import cast

from packaging.requirements import Requirement

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.engine.collection import DeduplicatedCollection
from pants.engine.env_vars import CompleteEnvironmentVars
from pants.engine.environment import EnvironmentName
from pants.engine.internals.selectors import Params
from pants.engine.internals.session import SessionValues
from pants.engine.process import ProcessCacheScope, 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


@rule
async def resolve_plugins(
    request: PluginsRequest,
    global_options: GlobalOptions,
) -> ResolvedPluginDistributions:
    """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.
    """
    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 ResolvedPluginDistributions()

    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 ResolvedPluginDistributions(plugins_process_result.stdout.decode().strip().split("\n"))


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(),
    ]

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	1✔
5
6	import importlib.metadata	1✔
7	import logging	1✔
8	import site	1✔
9	import sys	1✔
10	from collections.abc import Iterable	1✔
11	from dataclasses import dataclass	1✔
12	from typing import cast	1✔
13
14	from packaging.requirements import Requirement	1✔
15
16	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	1✔
17	from pants.backend.python.util_rules.pex import PexRequest, VenvPexProcess, create_venv_pex	1✔
18	from pants.backend.python.util_rules.pex_environment import PythonExecutable	1✔
19	from pants.backend.python.util_rules.pex_requirements import PexRequirements	1✔
20	from pants.core.environments.rules import determine_bootstrap_environment	1✔
21	from pants.engine.collection import DeduplicatedCollection	1✔
22	from pants.engine.env_vars import CompleteEnvironmentVars	1✔
23	from pants.engine.environment import EnvironmentName	1✔
24	from pants.engine.internals.selectors import Params	1✔
25	from pants.engine.internals.session import SessionValues	1✔
26	from pants.engine.process import ProcessCacheScope, execute_process_or_raise	1✔
27	from pants.engine.rules import QueryRule, collect_rules, implicitly, rule	1✔
28	from pants.init.bootstrap_scheduler import BootstrapScheduler	1✔
29	from pants.init.import_util import find_matching_distributions	1✔
30	from pants.option.global_options import GlobalOptions	1✔
31	from pants.option.options_bootstrapper import OptionsBootstrapper	1✔
32	from pants.util.logging import LogLevel	1✔
33
34	logger = logging.getLogger(__name__)	1✔
35
36
37	@dataclass(frozen=True)	1✔
38	class PluginsRequest:	1✔
39	# Interpreter constraints to resolve for, or None to resolve for the interpreter that Pants is
40	# running under.
41	interpreter_constraints: InterpreterConstraints \| None	1✔
42	# Requirement constraints to resolve with. If plugins will be loaded into the global working_set
43	# (i.e., onto the `sys.path`), then these should be the current contents of the working_set.
44	constraints: tuple[Requirement, ...]	1✔
45	# Backend requirements to resolve
46	requirements: tuple[str, ...]	1✔
47
48
49	class ResolvedPluginDistributions(DeduplicatedCollection[str]):	1✔
50	sort_input = True	1✔
51
52
53	@rule	1✔
54	async def resolve_plugins(	1✔
55	request: PluginsRequest,
56	global_options: GlobalOptions,
57	) -> ResolvedPluginDistributions:
58	"""This rule resolves plugins using a VenvPex, and exposes the absolute paths of their dists.
59
60	NB: This relies on the fact that PEX constructs venvs in a stable location (within the
61	`named_caches` directory), but consequently needs to disable the process cache: see the
62	ProcessCacheScope reference in the body.
63	"""
64	req_strings = sorted(global_options.plugins + request.requirements)	×
65
66	requirements = PexRequirements(	×
67	req_strings_or_addrs=req_strings,
68	constraints_strings=(str(constraint) for constraint in request.constraints),
69	description_of_origin="configured Pants plugins",
70	)
71	if not requirements:	×
72	return ResolvedPluginDistributions()	×
73
74	python: PythonExecutable \| None = None	×
75	if not request.interpreter_constraints:	×
76	python = PythonExecutable.fingerprinted(	×
77	sys.executable, ".".join(map(str, sys.version_info[:3])).encode("utf8")
78	)
79
80	plugins_pex = await create_venv_pex(	×
81	**implicitly(
82	PexRequest(
83	output_filename="pants_plugins.pex",
84	internal_only=True,
85	python=python,
86	requirements=requirements,
87	interpreter_constraints=request.interpreter_constraints or InterpreterConstraints(),
88	description=f"Resolving plugins: {', '.join(req_strings)}",
89	)
90	)
91	)
92
93	# NB: We run this Process per-restart because it (intentionally) leaks named cache
94	# paths in a way that invalidates the Process-cache. See the method doc.
95	cache_scope = (	×
96	ProcessCacheScope.PER_SESSION
97	if global_options.plugins_force_resolve
98	else ProcessCacheScope.PER_RESTART_SUCCESSFUL
99	)
100
101	plugins_process_result = await execute_process_or_raise(	×
102	**implicitly(
103	VenvPexProcess(
104	plugins_pex,
105	argv=("-c", "import os, site; print(os.linesep.join(site.getsitepackages()))"),
106	description="Extracting plugin locations",
107	level=LogLevel.DEBUG,
108	cache_scope=cache_scope,
109	)
110	)
111	)
112	return ResolvedPluginDistributions(plugins_process_result.stdout.decode().strip().split("\n"))	×
113
114
115	class PluginResolver:	1✔
116	"""Encapsulates the state of plugin loading.
117
118	Plugin loading is inherently stateful, and so the system enviroment on `sys.path` will be
119	mutated by each call to `PluginResolver.resolve`.
120	"""
121
122	def __init__(	1✔
123	self,
124	scheduler: BootstrapScheduler,
125	interpreter_constraints: InterpreterConstraints \| None = None,
126	inherit_existing_constraints: bool = True,
127	) -> None:
UNCOV 128	self._scheduler = scheduler	×
UNCOV 129	self._interpreter_constraints = interpreter_constraints	×
UNCOV 130	self._inherit_existing_constraints = inherit_existing_constraints	×
131
132	def resolve(	1✔
133	self,
134	options_bootstrapper: OptionsBootstrapper,
135	env: CompleteEnvironmentVars,
136	requirements: Iterable[str] = (),
137	) -> list[str]:
138	"""Resolves any configured plugins and adds them to the sys.path as a side effect."""
139
UNCOV 140	def to_requirement(d):	×
UNCOV 141	return f"{d.name}=={d.version}"	×
142
UNCOV 143	distributions: list[importlib.metadata.Distribution] = []	×
UNCOV 144	if self._inherit_existing_constraints:	×
UNCOV 145	distributions = list(find_matching_distributions(None))	×
146
UNCOV 147	request = PluginsRequest(	×
148	self._interpreter_constraints,
149	tuple(to_requirement(dist) for dist in distributions),
150	tuple(requirements),
151	)
152
UNCOV 153	result = []	×
UNCOV 154	for resolved_plugin_location in self._resolve_plugins(options_bootstrapper, env, request):	×
155	# Activate any .pth files plugin wheels may have.
UNCOV 156	orig_sys_path_len = len(sys.path)	×
UNCOV 157	site.addsitedir(resolved_plugin_location)	×
UNCOV 158	if len(sys.path) > orig_sys_path_len:	×
UNCOV 159	result.append(resolved_plugin_location)	×
160
UNCOV 161	return result	×
162
163	def _resolve_plugins(	1✔
164	self,
165	options_bootstrapper: OptionsBootstrapper,
166	env: CompleteEnvironmentVars,
167	request: PluginsRequest,
168	) -> ResolvedPluginDistributions:
UNCOV 169	session = self._scheduler.scheduler.new_session(	×
170	"plugin_resolver",
171	session_values=SessionValues(
172	{
173	OptionsBootstrapper: options_bootstrapper,
174	CompleteEnvironmentVars: env,
175	}
176	),
177	)
UNCOV 178	params = Params(request, determine_bootstrap_environment(session))	×
UNCOV 179	return cast(	×
180	ResolvedPluginDistributions,
181	session.product_request(ResolvedPluginDistributions, [params])[0],
182	)
183
184
185	def rules():	1✔
UNCOV 186	return [	×
187	QueryRule(ResolvedPluginDistributions, [PluginsRequest, EnvironmentName]),
188	*collect_rules(),
189	]

pantsbuild / pants / 19015773527

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