21552830208

Committed 31 Jan 2026 11:40PM UTC coverage: 80.277% (-0.05%) from 80.324%

Build # 21552830208

Build Type

Pull #23062

github

Committed by

web-flow

Commit Message

Merge 808a9786c into 2c4dcf9cf

Pull Request Pull Request #23062: Remove support for Get

Run Details

18 of 25 new or added lines in 4 files covered. (72.0%)

17119 existing lines in 541 files now uncovered.

78278 of 97510 relevant lines covered (80.28%)

3.36 hits per line

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

77.63

/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	12✔
5
6	import dataclasses	12✔
7	import importlib	12✔
8	import logging	12✔
9	import sys	12✔
10	from collections.abc import Iterator	12✔
11	from contextlib import contextmanager	12✔
12	from pathlib import Path	12✔
13
14	from pants.build_graph.build_configuration import BuildConfiguration	12✔
15	from pants.engine.env_vars import CompleteEnvironmentVars	12✔
16	from pants.engine.internals.native_engine import PyExecutor	12✔
17	from pants.engine.unions import UnionMembership	12✔
18	from pants.help.flag_error_help_printer import FlagErrorHelpPrinter	12✔
19	from pants.init.bootstrap_scheduler import BootstrapScheduler	12✔
20	from pants.init.engine_initializer import EngineInitializer	12✔
21	from pants.init.extension_loader import (	12✔
22	load_backends_and_plugins,
23	load_build_configuration_from_source,
24	)
25	from pants.init.plugin_resolver import PluginResolver	12✔
26	from pants.init.plugin_resolver import rules as plugin_resolver_rules	12✔
27	from pants.option.bootstrap_options import DynamicRemoteOptions	12✔
28	from pants.option.errors import UnknownFlagsError	12✔
29	from pants.option.options import Options	12✔
30	from pants.option.options_bootstrapper import OptionsBootstrapper	12✔
31	from pants.util.requirements import parse_requirements_file	12✔
32
33	logger = logging.getLogger(__name__)	12✔
34
35
36	def _initialize_build_configuration(	12✔
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
UNCOV 48	bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()	4✔
49
50	# Add any extra paths to python path (e.g., for loading extra source backends).
UNCOV 51	for path in bootstrap_options.pythonpath:	4✔
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.
UNCOV 57	backends_requirements = _collect_backends_requirements(bootstrap_options.backend_packages)	4✔
UNCOV 58	plugin_resolver.resolve(options_bootstrapper, env, backends_requirements)	4✔
59
60	# Load plugins and backends.
UNCOV 61	return load_backends_and_plugins(	3✔
62	bootstrap_options.plugins,
63	bootstrap_options.backend_packages,
64	)
65
66
67	def _collect_backends_requirements(backends: list[str]) -> list[str]:	12✔
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	"""
UNCOV 74	requirements = []	4✔
75
UNCOV 76	for backend_package in backends:	4✔
UNCOV 77	try:	1✔
UNCOV 78	backend_package_spec = importlib.util.find_spec(backend_package)	1✔
79	except ModuleNotFoundError:	×
80	continue	×
81
UNCOV 82	if backend_package_spec is None:	1✔
83	continue	×
84
UNCOV 85	if backend_package_spec.origin is None:	1✔
UNCOV 86	logger.warning(	1✔
87	f"Can not check requirements for backend: '{backend_package}'. A __init__.py file is probably missing."
88	)
UNCOV 89	continue	1✔
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
UNCOV 102	return requirements	4✔
103
104
105	def create_bootstrap_scheduler(	12✔
106	options_bootstrapper: OptionsBootstrapper, executor: PyExecutor
107	) -> BootstrapScheduler:
UNCOV 108	bc_builder = BuildConfiguration.Builder()	4✔
109	# To load plugins, we only need access to the Python/PEX rules.
UNCOV 110	load_build_configuration_from_source(bc_builder, ["pants.backend.python"])	4✔
111	# And to plugin-loading-specific rules.
UNCOV 112	bc_builder.register_rules("_dummy_for_bootstrapping_", plugin_resolver_rules())	4✔
113	# We allow unrecognized options to defer any option error handling until post-bootstrap.
UNCOV 114	bc_builder.allow_unknown_options()	4✔
UNCOV 115	return BootstrapScheduler(	4✔
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:	12✔
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__(	12✔
140	self,
141	options_bootstrapper: OptionsBootstrapper,
142	executor: PyExecutor,
143	) -> None:
UNCOV 144	self._bootstrap_scheduler = create_bootstrap_scheduler(options_bootstrapper, executor)	4✔
UNCOV 145	self._plugin_resolver = PluginResolver(self._bootstrap_scheduler)	4✔
146
147	def build_config(	12✔
148	self,
149	options_bootstrapper: OptionsBootstrapper,
150	env: CompleteEnvironmentVars,
151	) -> BuildConfiguration:
UNCOV 152	return _initialize_build_configuration(self._plugin_resolver, options_bootstrapper, env)	4✔
153
154	def options(	12✔
155	self,
156	options_bootstrapper: OptionsBootstrapper,
157	env: CompleteEnvironmentVars,
158	build_config: BuildConfiguration,
159	union_membership: UnionMembership,
160	*,
161	raise_: bool,
162	) -> Options:
UNCOV 163	with self.handle_unknown_flags(options_bootstrapper, env, raise_=raise_):	3✔
UNCOV 164	return options_bootstrapper.full_options(	3✔
165	build_config.known_scope_infos, union_membership, build_config.allow_unknown_options
166	)
167
168	@contextmanager	12✔
169	def handle_unknown_flags(	12✔
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."""
UNCOV 177	try:	3✔
UNCOV 178	yield	3✔
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 / 21552830208

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