20632486505

Committed 01 Jan 2026 04:21AM UTC coverage: 43.231% (-37.1%) from 80.281%

Build # 20632486505

Build Type

Pull #22962

github

Committed by

web-flow

Commit Message

Merge 08d5c63b0 into f52ab6675

Pull Request Pull Request #22962: Bump the gha-deps group across 1 directory with 6 updates

Run Details

26122 of 60424 relevant lines covered (43.23%)

0.86 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(),
                additional_args=("--preserve-pip-download-log", "pex-pip-download.log"),
                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	2✔
5
6	import importlib.metadata	2✔
7	import logging	2✔
8	import site	2✔
9	import sys	2✔
10	from collections.abc import Iterable	2✔
11	from dataclasses import dataclass	2✔
12	from typing import cast	2✔
13
14	from packaging.requirements import Requirement	2✔
15
16	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	2✔
17	from pants.backend.python.util_rules.pex import PexRequest, VenvPexProcess, create_venv_pex	2✔
18	from pants.backend.python.util_rules.pex_environment import PythonExecutable	2✔
19	from pants.backend.python.util_rules.pex_requirements import PexRequirements	2✔
20	from pants.core.environments.rules import determine_bootstrap_environment	2✔
21	from pants.engine.collection import DeduplicatedCollection	2✔
22	from pants.engine.env_vars import CompleteEnvironmentVars	2✔
23	from pants.engine.environment import EnvironmentName	2✔
24	from pants.engine.internals.selectors import Params	2✔
25	from pants.engine.internals.session import SessionValues	2✔
26	from pants.engine.process import ProcessCacheScope, execute_process_or_raise	2✔
27	from pants.engine.rules import QueryRule, collect_rules, implicitly, rule	2✔
28	from pants.init.bootstrap_scheduler import BootstrapScheduler	2✔
29	from pants.init.import_util import find_matching_distributions	2✔
30	from pants.option.global_options import GlobalOptions	2✔
31	from pants.option.options_bootstrapper import OptionsBootstrapper	2✔
32	from pants.util.logging import LogLevel	2✔
33
34	logger = logging.getLogger(__name__)	2✔
35
36
37	@dataclass(frozen=True)	2✔
38	class PluginsRequest:	2✔
39	# Interpreter constraints to resolve for, or None to resolve for the interpreter that Pants is
40	# running under.
41	interpreter_constraints: InterpreterConstraints \| None	2✔
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, ...]	2✔
45	# Backend requirements to resolve
46	requirements: tuple[str, ...]	2✔
47
48
49	class ResolvedPluginDistributions(DeduplicatedCollection[str]):	2✔
50	sort_input = True	2✔
51
52
53	@rule	2✔
54	async def resolve_plugins(	2✔
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	additional_args=("--preserve-pip-download-log", "pex-pip-download.log"),
89	description=f"Resolving plugins: {', '.join(req_strings)}",
90	)
91	)
92	)
93
94	# NB: We run this Process per-restart because it (intentionally) leaks named cache
95	# paths in a way that invalidates the Process-cache. See the method doc.
96	cache_scope = (	×
97	ProcessCacheScope.PER_SESSION
98	if global_options.plugins_force_resolve
99	else ProcessCacheScope.PER_RESTART_SUCCESSFUL
100	)
101
102	plugins_process_result = await execute_process_or_raise(	×
103	**implicitly(
104	VenvPexProcess(
105	plugins_pex,
106	argv=("-c", "import os, site; print(os.linesep.join(site.getsitepackages()))"),
107	description="Extracting plugin locations",
108	level=LogLevel.DEBUG,
109	cache_scope=cache_scope,
110	)
111	)
112	)
113	return ResolvedPluginDistributions(plugins_process_result.stdout.decode().strip().split("\n"))	×
114
115
116	class PluginResolver:	2✔
117	"""Encapsulates the state of plugin loading.
118
119	Plugin loading is inherently stateful, and so the system enviroment on `sys.path` will be
120	mutated by each call to `PluginResolver.resolve`.
121	"""
122
123	def __init__(	2✔
124	self,
125	scheduler: BootstrapScheduler,
126	interpreter_constraints: InterpreterConstraints \| None = None,
127	inherit_existing_constraints: bool = True,
128	) -> None:
129	self._scheduler = scheduler	×
130	self._interpreter_constraints = interpreter_constraints	×
131	self._inherit_existing_constraints = inherit_existing_constraints	×
132
133	def resolve(	2✔
134	self,
135	options_bootstrapper: OptionsBootstrapper,
136	env: CompleteEnvironmentVars,
137	requirements: Iterable[str] = (),
138	) -> list[str]:
139	"""Resolves any configured plugins and adds them to the sys.path as a side effect."""
140
141	def to_requirement(d):	×
142	return f"{d.name}=={d.version}"	×
143
144	distributions: list[importlib.metadata.Distribution] = []	×
145	if self._inherit_existing_constraints:	×
146	distributions = list(find_matching_distributions(None))	×
147
148	request = PluginsRequest(	×
149	self._interpreter_constraints,
150	tuple(to_requirement(dist) for dist in distributions),
151	tuple(requirements),
152	)
153
154	result = []	×
155	for resolved_plugin_location in self._resolve_plugins(options_bootstrapper, env, request):	×
156	# Activate any .pth files plugin wheels may have.
157	orig_sys_path_len = len(sys.path)	×
158	site.addsitedir(resolved_plugin_location)	×
159	if len(sys.path) > orig_sys_path_len:	×
160	result.append(resolved_plugin_location)	×
161
162	return result	×
163
164	def _resolve_plugins(	2✔
165	self,
166	options_bootstrapper: OptionsBootstrapper,
167	env: CompleteEnvironmentVars,
168	request: PluginsRequest,
169	) -> ResolvedPluginDistributions:
170	session = self._scheduler.scheduler.new_session(	×
171	"plugin_resolver",
172	session_values=SessionValues(
173	{
174	OptionsBootstrapper: options_bootstrapper,
175	CompleteEnvironmentVars: env,
176	}
177	),
178	)
179	params = Params(request, determine_bootstrap_environment(session))	×
180	return cast(	×
181	ResolvedPluginDistributions,
182	session.product_request(ResolvedPluginDistributions, params)[0],
183	)
184
185
186	def rules():	2✔
187	return [	×
188	QueryRule(ResolvedPluginDistributions, [PluginsRequest, EnvironmentName]),
189	*collect_rules(),
190	]

pantsbuild / pants / 20632486505

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