21038851336

Committed 15 Jan 2026 04:37PM UTC coverage: 78.666% (-1.6%) from 80.269%

Build # 21038851336

Build Type

push

github

Committed by

web-flow

Commit Message

update to Pex 2.81.0 (#23018)

Release Notes:
 * https://github.com/pex-tool/pex/releases/tag/v2.80.0
 * https://github.com/pex-tool/pex/releases/tag/v2.81.0

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

1077 existing lines in 52 files now uncovered.

75010 of 95353 relevant lines covered (78.67%)

3.17 hits per line

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

69.74

/src/python/pants/init/options_initializer.py

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

from __future__ import annotations

import dataclasses
import importlib
import logging
import sys
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path

from pants.build_graph.build_configuration import BuildConfiguration
from pants.engine.env_vars import CompleteEnvironmentVars
from pants.engine.internals.native_engine import PyExecutor
from pants.engine.unions import UnionMembership
from pants.help.flag_error_help_printer import FlagErrorHelpPrinter
from pants.init.bootstrap_scheduler import BootstrapScheduler
from pants.init.engine_initializer import EngineInitializer
from pants.init.extension_loader import (
    load_backends_and_plugins,
    load_build_configuration_from_source,
)
from pants.init.plugin_resolver import PluginResolver
from pants.init.plugin_resolver import rules as plugin_resolver_rules
from pants.option.bootstrap_options import DynamicRemoteOptions
from pants.option.errors import UnknownFlagsError
from pants.option.options import Options
from pants.option.options_bootstrapper import OptionsBootstrapper
from pants.util.requirements import parse_requirements_file

logger = logging.getLogger(__name__)


def _initialize_build_configuration(
    plugin_resolver: PluginResolver,
    options_bootstrapper: OptionsBootstrapper,
    env: CompleteEnvironmentVars,
) -> BuildConfiguration:
    """Initialize a BuildConfiguration for the given OptionsBootstrapper.

    NB: This method:
      1. has the side effect of (idempotently) adding PYTHONPATH entries for this process
      2. is expensive to call, because it might resolve plugins from the network
    """

    bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()

    # Add any extra paths to python path (e.g., for loading extra source backends).
    for path in bootstrap_options.pythonpath:
        if path not in sys.path:
            sys.path.append(path)

    # Resolve the actual Python code for any plugins. `sys.path` is modified as a side effect if
    # plugins were configured.
    backends_requirements = _collect_backends_requirements(bootstrap_options.backend_packages)
    plugin_resolver.resolve(options_bootstrapper, env, backends_requirements)

    # Load plugins and backends.
    return load_backends_and_plugins(
        bootstrap_options.plugins,
        bootstrap_options.backend_packages,
    )


def _collect_backends_requirements(backends: list[str]) -> list[str]:
    """Collects backend package dependencies, in case those are declared in an adjacent
    requirements.txt. Ignores any loading errors, assuming those will be later on handled by the
    backends loader.

    :param backends: An list of packages to load v2 backends requirements from.
    """
    requirements = []

    for backend_package in backends:
        try:
            backend_package_spec = importlib.util.find_spec(backend_package)
        except ModuleNotFoundError:
            continue

        if backend_package_spec is None:
            continue

        if backend_package_spec.origin is None:
            logger.warning(
                f"Can not check requirements for backend: '{backend_package}'. A __init__.py file is probably missing."
            )
            continue

        requirements_txt_file_path = Path(backend_package_spec.origin).parent.joinpath(
            "requirements.txt"
        )
        if requirements_txt_file_path.exists():
            content = requirements_txt_file_path.read_text()
            backend_package_requirements = [
                str(r)
                for r in parse_requirements_file(content, rel_path=str(requirements_txt_file_path))
            ]
            requirements.extend(backend_package_requirements)

    return requirements


def create_bootstrap_scheduler(
    options_bootstrapper: OptionsBootstrapper, executor: PyExecutor
) -> BootstrapScheduler:
    bc_builder = BuildConfiguration.Builder()
    # To load plugins, we only need access to the Python/PEX rules.
    load_build_configuration_from_source(bc_builder, ["pants.backend.python"])
    # And to plugin-loading-specific rules.
    bc_builder.register_rules("_dummy_for_bootstrapping_", plugin_resolver_rules())
    # We allow unrecognized options to defer any option error handling until post-bootstrap.
    bc_builder.allow_unknown_options()
    return BootstrapScheduler(
        EngineInitializer.setup_graph(
            options_bootstrapper.bootstrap_options.for_global_scope(),
            bc_builder.create(),
            DynamicRemoteOptions.disabled(),
            executor,
            is_bootstrap=True,
        ).scheduler
    )


class OptionsInitializer:
    """Initializes BuildConfiguration and Options instances given an OptionsBootstrapper.

    NB: Although this constructor takes an instance of the OptionsBootstrapper, it is
    used only to construct a "bootstrap" Scheduler: actual calls to resolve plugins use a
    per-request instance of the OptionsBootstrapper, which might request different plugins.

    TODO: We would eventually like to use the bootstrap Scheduler to construct the
    OptionsBootstrapper as well, but for now we do the opposite thing, and the Scheduler is
    used only to resolve plugins.
      see https://github.com/pantsbuild/pants/pull/11568
    """

    def __init__(
        self,
        options_bootstrapper: OptionsBootstrapper,
        executor: PyExecutor,
    ) -> None:
        self._bootstrap_scheduler = create_bootstrap_scheduler(options_bootstrapper, executor)
        self._plugin_resolver = PluginResolver(self._bootstrap_scheduler)

    def build_config(
        self,
        options_bootstrapper: OptionsBootstrapper,
        env: CompleteEnvironmentVars,
    ) -> BuildConfiguration:
        return _initialize_build_configuration(self._plugin_resolver, options_bootstrapper, env)

    def options(
        self,
        options_bootstrapper: OptionsBootstrapper,
        env: CompleteEnvironmentVars,
        build_config: BuildConfiguration,
        union_membership: UnionMembership,
        *,
        raise_: bool,
    ) -> Options:
        with self.handle_unknown_flags(options_bootstrapper, env, raise_=raise_):
            return options_bootstrapper.full_options(
                build_config.known_scope_infos, union_membership, build_config.allow_unknown_options
            )

    @contextmanager
    def handle_unknown_flags(
        self,
        options_bootstrapper: OptionsBootstrapper,
        env: CompleteEnvironmentVars,
        *,
        raise_: bool,
    ) -> Iterator[None]:
        """If there are any unknown flags, print "Did you mean?" and possibly error."""
        try:
            yield
        except UnknownFlagsError as err:
            build_config = _initialize_build_configuration(
                self._plugin_resolver, options_bootstrapper, env
            )
            # We need an options instance in order to get "did you mean" suggestions, but we know
            # there are bad flags in the args, so we generate options with no flags.
            no_arg_bootstrapper = dataclasses.replace(
                options_bootstrapper, args=("dummy_first_arg",)
            )
            options = no_arg_bootstrapper.full_options(
                build_config.known_scope_infos,
                union_membership=UnionMembership.empty(),
            )
            FlagErrorHelpPrinter(options).handle_unknown_flags(err)
            if raise_:
                raise err

1	# Copyright 2016 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	11✔
5
6	import dataclasses	11✔
7	import importlib	11✔
8	import logging	11✔
9	import sys	11✔
10	from collections.abc import Iterator	11✔
11	from contextlib import contextmanager	11✔
12	from pathlib import Path	11✔
13
14	from pants.build_graph.build_configuration import BuildConfiguration	11✔
15	from pants.engine.env_vars import CompleteEnvironmentVars	11✔
16	from pants.engine.internals.native_engine import PyExecutor	11✔
17	from pants.engine.unions import UnionMembership	11✔
18	from pants.help.flag_error_help_printer import FlagErrorHelpPrinter	11✔
19	from pants.init.bootstrap_scheduler import BootstrapScheduler	11✔
20	from pants.init.engine_initializer import EngineInitializer	11✔
21	from pants.init.extension_loader import (	11✔
22	load_backends_and_plugins,
23	load_build_configuration_from_source,
24	)
25	from pants.init.plugin_resolver import PluginResolver	11✔
26	from pants.init.plugin_resolver import rules as plugin_resolver_rules	11✔
27	from pants.option.bootstrap_options import DynamicRemoteOptions	11✔
28	from pants.option.errors import UnknownFlagsError	11✔
29	from pants.option.options import Options	11✔
30	from pants.option.options_bootstrapper import OptionsBootstrapper	11✔
31	from pants.util.requirements import parse_requirements_file	11✔
32
33	logger = logging.getLogger(__name__)	11✔
34
35
36	def _initialize_build_configuration(	11✔
37	plugin_resolver: PluginResolver,
38	options_bootstrapper: OptionsBootstrapper,
39	env: CompleteEnvironmentVars,
40	) -> BuildConfiguration:
41	"""Initialize a BuildConfiguration for the given OptionsBootstrapper.
42
43	NB: This method:
44	1. has the side effect of (idempotently) adding PYTHONPATH entries for this process
45	2. is expensive to call, because it might resolve plugins from the network
46	"""
47
48	bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()	3✔
49
50	# Add any extra paths to python path (e.g., for loading extra source backends).
51	for path in bootstrap_options.pythonpath:	3✔
52	if path not in sys.path:	×
53	sys.path.append(path)	×
54
55	# Resolve the actual Python code for any plugins. `sys.path` is modified as a side effect if
56	# plugins were configured.
57	backends_requirements = _collect_backends_requirements(bootstrap_options.backend_packages)	3✔
58	plugin_resolver.resolve(options_bootstrapper, env, backends_requirements)	3✔
59
60	# Load plugins and backends.
61	return load_backends_and_plugins(	2✔
62	bootstrap_options.plugins,
63	bootstrap_options.backend_packages,
64	)
65
66
67	def _collect_backends_requirements(backends: list[str]) -> list[str]:	11✔
68	"""Collects backend package dependencies, in case those are declared in an adjacent
69	requirements.txt. Ignores any loading errors, assuming those will be later on handled by the
70	backends loader.
71
72	:param backends: An list of packages to load v2 backends requirements from.
73	"""
74	requirements = []	3✔
75
76	for backend_package in backends:	3✔
UNCOV 77	try:	×
UNCOV 78	backend_package_spec = importlib.util.find_spec(backend_package)	×
79	except ModuleNotFoundError:	×
80	continue	×
81
UNCOV 82	if backend_package_spec is None:	×
83	continue	×
84
UNCOV 85	if backend_package_spec.origin is None:	×
UNCOV 86	logger.warning(	×
87	f"Can not check requirements for backend: '{backend_package}'. A __init__.py file is probably missing."
88	)
UNCOV 89	continue	×
90
91	requirements_txt_file_path = Path(backend_package_spec.origin).parent.joinpath(	×
92	"requirements.txt"
93	)
94	if requirements_txt_file_path.exists():	×
95	content = requirements_txt_file_path.read_text()	×
96	backend_package_requirements = [	×
97	str(r)
98	for r in parse_requirements_file(content, rel_path=str(requirements_txt_file_path))
99	]
100	requirements.extend(backend_package_requirements)	×
101
102	return requirements	3✔
103
104
105	def create_bootstrap_scheduler(	11✔
106	options_bootstrapper: OptionsBootstrapper, executor: PyExecutor
107	) -> BootstrapScheduler:
108	bc_builder = BuildConfiguration.Builder()	3✔
109	# To load plugins, we only need access to the Python/PEX rules.
110	load_build_configuration_from_source(bc_builder, ["pants.backend.python"])	3✔
111	# And to plugin-loading-specific rules.
112	bc_builder.register_rules("_dummy_for_bootstrapping_", plugin_resolver_rules())	3✔
113	# We allow unrecognized options to defer any option error handling until post-bootstrap.
114	bc_builder.allow_unknown_options()	3✔
115	return BootstrapScheduler(	3✔
116	EngineInitializer.setup_graph(
117	options_bootstrapper.bootstrap_options.for_global_scope(),
118	bc_builder.create(),
119	DynamicRemoteOptions.disabled(),
120	executor,
121	is_bootstrap=True,
122	).scheduler
123	)
124
125
126	class OptionsInitializer:	11✔
127	"""Initializes BuildConfiguration and Options instances given an OptionsBootstrapper.
128
129	NB: Although this constructor takes an instance of the OptionsBootstrapper, it is
130	used only to construct a "bootstrap" Scheduler: actual calls to resolve plugins use a
131	per-request instance of the OptionsBootstrapper, which might request different plugins.
132
133	TODO: We would eventually like to use the bootstrap Scheduler to construct the
134	OptionsBootstrapper as well, but for now we do the opposite thing, and the Scheduler is
135	used only to resolve plugins.
136	see https://github.com/pantsbuild/pants/pull/11568
137	"""
138
139	def __init__(	11✔
140	self,
141	options_bootstrapper: OptionsBootstrapper,
142	executor: PyExecutor,
143	) -> None:
144	self._bootstrap_scheduler = create_bootstrap_scheduler(options_bootstrapper, executor)	3✔
145	self._plugin_resolver = PluginResolver(self._bootstrap_scheduler)	3✔
146
147	def build_config(	11✔
148	self,
149	options_bootstrapper: OptionsBootstrapper,
150	env: CompleteEnvironmentVars,
151	) -> BuildConfiguration:
152	return _initialize_build_configuration(self._plugin_resolver, options_bootstrapper, env)	3✔
153
154	def options(	11✔
155	self,
156	options_bootstrapper: OptionsBootstrapper,
157	env: CompleteEnvironmentVars,
158	build_config: BuildConfiguration,
159	union_membership: UnionMembership,
160	*,
161	raise_: bool,
162	) -> Options:
163	with self.handle_unknown_flags(options_bootstrapper, env, raise_=raise_):	2✔
164	return options_bootstrapper.full_options(	2✔
165	build_config.known_scope_infos, union_membership, build_config.allow_unknown_options
166	)
167
168	@contextmanager	11✔
169	def handle_unknown_flags(	11✔
170	self,
171	options_bootstrapper: OptionsBootstrapper,
172	env: CompleteEnvironmentVars,
173	*,
174	raise_: bool,
175	) -> Iterator[None]:
176	"""If there are any unknown flags, print "Did you mean?" and possibly error."""
177	try:	2✔
178	yield	2✔
179	except UnknownFlagsError as err:	×
180	build_config = _initialize_build_configuration(	×
181	self._plugin_resolver, options_bootstrapper, env
182	)
183	# We need an options instance in order to get "did you mean" suggestions, but we know
184	# there are bad flags in the args, so we generate options with no flags.
185	no_arg_bootstrapper = dataclasses.replace(	×
186	options_bootstrapper, args=("dummy_first_arg",)
187	)
188	options = no_arg_bootstrapper.full_options(	×
189	build_config.known_scope_infos,
190	union_membership=UnionMembership.empty(),
191	)
192	FlagErrorHelpPrinter(options).handle_unknown_flags(err)	×
193	if raise_:	×
194	raise err	×

pantsbuild / pants / 21038851336

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