23274822986

Committed 19 Mar 2026 01:06AM UTC coverage: 92.926% (-0.006%) from 92.932%

Build # 23274822986

Build Type

push

github

Committed by

web-flow

Commit Message

[pants_ng] An NG subsystem implementation (#23165)

Like OG subsystems, these are containers of options
read from flags, env vars and config files.

Unlike OG subsystems, these are not necessarily
singletons - you can have multiple instances of the
same subsystem active for different partitions of
sources that have different local config files.

Making OG subsystems non-singleton was not feasible,
for both design and performance reasons.

These NG subsystems are lighter weight and more 
streamlined than their OG counterparts:

- They support a more limited set of value types.
- Option registration is via a new `@option` decorator
  instead of the large menagerie of field type classes.

AI disclosure: Claude found a way to create a mypy plugin
  that prevents spurious empty body errors. I had to manually
  tweak the plugin to make it less verbose and more streamlined.

Run Details

269 of 295 new or added lines in 2 files covered. (91.19%)

1 existing line in 1 file now uncovered.

91440 of 98401 relevant lines covered (92.93%)

4.04 hits per line

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

85.63

/src/python/pants/ng/subsystem.py

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

import functools
import inspect
import typing
from collections.abc import Callable
from dataclasses import dataclass
from functools import partial, wraps
from types import GenericAlias
from typing import Any, Iterable, Self, Tuple, TypeVar, cast

from pants.engine.internals.native_engine import PyNgOptions, PyNgOptionsReader, PyOptionId
from pants.engine.rules import Rule, TaskRule, collect_rules, rule
from pants.ng.source_partition import SourcePartition
from pants.option.ranked_value import Rank
from pants.util.frozendict import FrozenDict
from pants.util.memo import memoized_classmethod

# Supported option value types.
#
# Note that this is a subset of the value types supported by og Pants, which allows
# nested and heterogenously-valued dicts. However that doesn't seem to be worth the effort
# so let's see how far we can get without supporting those. Note also that these types are all
# immutable, whereas the og types did not have to be. Again, not a complication we should
# be enthusiastic to support.
ScalarValue = bool | str | int | float
OptionValue = ScalarValue | tuple[ScalarValue, ...] | FrozenDict[str, str]


_SCALAR_TYPES = typing.get_args(ScalarValue)


def _is_valid_type(t: type | GenericAlias) -> bool:
    if isinstance(t, type):
        # ScalarType
        return t in _SCALAR_TYPES

    torig = typing.get_origin(t)
    targs = typing.get_args(t)
    return (
        # tuple[ScalarType, ...]
        torig == tuple and len(targs) == 2 and targs[0] in _SCALAR_TYPES and targs[1] == Ellipsis
    ) or (
        # FrozenDict[str, str]
        torig == FrozenDict and targs == (str, str)
    )


def _type_to_readable_str(t: type | GenericAlias) -> str:
    if isinstance(t, type):
        # E.g., `str` instead of `<class 'str'>`
        return t.__name__
    # A GenericAlias renders as you'd expect (e.g., `tuple[str, ...]`).
    return str(t)


def _is_of_type(val: OptionValue, expected_type: type | GenericAlias) -> bool:
    if isinstance(expected_type, type) and expected_type in _SCALAR_TYPES:
        return isinstance(val, expected_type)

    torig = typing.get_origin(expected_type)
    targs = typing.get_args(expected_type)
    if torig == tuple:
        member_type = targs[0]
        return isinstance(val, tuple) and all(isinstance(member, member_type) for member in val)
    if torig == FrozenDict:
        return isinstance(val, FrozenDict) and all(
            (type(k), type(v)) == (str, str) for k, v in val.items()
        )
    # Should never happen, assuming we've already passed _is_valid_type().
    raise ValueError(f"Unexpected expected_type {expected_type}")


def _getter(getter: Callable, option_parser, option_id, default) -> tuple[Any, bool]:
    # The legacy getter returns a tuple of (value, rank, optional derivation),
    # we only care about value and rank.
    value, rank, _ = getter(option_parser, option_id, default)
    return value, rank > Rank.HARDCODED._rank


def _tuple_getter(
    getter: Callable, option_parser, option_id, default
) -> tuple[tuple[ScalarValue, ...], bool]:
    # The legacy getter returns a tuple of (list value, rank, optional derivation),
    # we only care about value and rank, and we convert the value to tuple. The rust code can
    # consume a tuple where a list is expected, so no need to convert the default.
    value, rank, _ = getter(option_parser, option_id, default or ())
    return tuple(value), rank > Rank.HARDCODED._rank


def _dict_getter(option_parser, option_id, default) -> tuple[FrozenDict, bool]:
    # The legacy getter returns a tuple of (value, rank, optional derivation),
    # we only care about value and rank, and we convert the value to FrozenDict.
    # We also convert the default to regular dict, so the Rust code can consume it.
    # The type checker currently doesn't grok the overloaded FrozenDict.__init__, so
    # we ignore the arg-type error.
    value, rank, _ = PyNgOptionsReader.get_dict(option_parser, option_id, dict(default or {}))
    return FrozenDict(value), rank > Rank.HARDCODED._rank  # type: ignore[arg-type]


# A map from type to getter for that type.
# Each getter is called with the arguments (options_reader, option_id, default) and
# returns a pair (value, is_explicit). is_explicit is True if the value was specified
# explicitly (by config, env var or flag), and False if the default was used.
# Note that this differentiates between the case of "no explicit value" vs "explicit value
# was provided, but it happened to be the same as the default value".
_getters: dict[type | GenericAlias, Callable] = {
    bool: partial(_getter, PyNgOptionsReader.get_bool),
    str: partial(_getter, PyNgOptionsReader.get_string),
    int: partial(_getter, PyNgOptionsReader.get_int),
    float: partial(_getter, PyNgOptionsReader.get_float),
    tuple[bool, ...]: partial(_tuple_getter, PyNgOptionsReader.get_bool_list),
    tuple[str, ...]: partial(_tuple_getter, PyNgOptionsReader.get_string_list),
    tuple[int, ...]: partial(_tuple_getter, PyNgOptionsReader.get_int_list),
    tuple[float, ...]: partial(_tuple_getter, PyNgOptionsReader.get_float_list),
    FrozenDict[str, str]: _dict_getter,
}


@dataclass(frozen=True)
class OptionDescriptor:
    """Information we capture at @option evaluation time."""

    name: str
    type: type | GenericAlias
    default: OptionValue | None
    help: str
    required: bool = False


R = TypeVar("R")
S = TypeVar("S")
OptionFunc = Callable[[S], R]
Default = OptionValue | Callable[[Any], OptionValue]
Help = str | Callable[[Any], str]


REQUIRED = object()


# A @decorator for options.
def option(*, help: Help, required: bool = False, default: Default | None = None) -> OptionFunc:
    def decorator(func: OptionFunc):
        # Mark the func as an option.
        func._option_ = True  # type: ignore[attr-defined]
        func._option_required_ = required  # type: ignore[attr-defined]
        func._option_default_ = default  # type: ignore[attr-defined]
        func._option_help_ = help  # type: ignore[attr-defined]

        @wraps(func)
        def wrapper(instance):
            return instance._option_values_[func.__name__]

        return property(wrapper)

    return decorator


class SubsystemNg:
    """A holder of options in a scope.

    Much lighter weight than its og counterpart.

    Use like this:

    class Foo(SubsystemNg):
        options_scope = "foo"

        @option(help="bar help")
        def bar(self) -> str: ...

        @option(default=42, help="baz help")
        def baz(self) -> int: ...

        ...

    The supported option types are: bool, str, int, float, tuple[bool, ...], tuple[str, ...],
    tuple[int, ...], tuple[float, ...] and FrozenDict[str, str].

    The `...` in the method bodies above are literal: the subsystem mechanism will provide the body.
    You may need `# mypy: disable-error-code=empty-body` for your code to pass mypy type checks, or
    you can use the mypy plugin at pants.ng.subsystem_mypy_plugin to suppress those errors.

    `default` and `help` can be values or callables that take the subsystem type as an arg, and
    return the relevant values.

    Subsystem instances are instantiated with a PyNgOptionsReader, and calling .bar or .baz (with
    no parentheses) on an instance of Foo will return the value of that option from the underlying
    parser.

    Option names must not _begin_and_end_ with an underscore. Such names are reserved for the
    underlying initialization mechanics.
    """

    _subsystem_ng_ = True

    # Subclasses must set.
    options_scope: str | None = None
    help: str | None = None

    # The ctor will set.
    _option_values_: FrozenDict[str, OptionValue] | None = None

    @classmethod
    def create(cls, options_reader: PyNgOptionsReader) -> Self:
        """Indirect constructor method.

        Useful for avoiding spurious mypy errors when instantiating instances in tests.
        """
        return cls(options_reader)

    def __init__(self, options_reader: PyNgOptionsReader):
        option_values = {}
        for descriptor in self._get_option_descriptors_():
            option_id = PyOptionId(descriptor.name, scope=self.options_scope)
            getter = _getters[descriptor.type]
            val, is_explicit = getter(
                options_reader, option_id=option_id, default=descriptor.default
            )
            if descriptor.required and not is_explicit:
                raise ValueError(
                    "No value provided for required option "
                    f"[{self.options_scope}].{descriptor.name}."
                )
            option_values[descriptor.name] = val
        self._option_values_ = FrozenDict(option_values)

    def _get_option_values_(self) -> FrozenDict[str, OptionValue]:
        return cast(FrozenDict[str, OptionValue], self._option_values_)

    def __str__(self) -> str:
        fields = [f"{k}={v}" for k, v in self._get_option_values_().items()]
        return f"{self.__class__.__name__}({', '.join(fields)})"

    @classmethod
    def _get_option_descriptors_(cls) -> tuple[OptionDescriptor, ...]:
        """Return the option descriptors for this subsystem.

        Must only be called after initialize() has been called.
        """
        return cast(tuple[OptionDescriptor], getattr(cls, "_option_descriptors_"))

    @classmethod
    def _initialize_(cls):
        if getattr(cls, "_option_descriptors_", None) is not None:
            return

        if cls.options_scope is None:
            raise ValueError(f"Subsystem class {cls.__name__} must set the options_scope classvar.")
        if cls.help is None:
            raise ValueError(f"Subsystem class {cls.__name__} must set the help classvar.")
        option_descriptors = []
        # We don't use dir() or inspect.getmembers() because those sort by name,
        # and we want to preserve declaration order.
        for basecls in inspect.getmro(cls):
            for name, member in basecls.__dict__.items():
                if not isinstance(member, property):
                    continue
                wrapped = member.fget
                if not getattr(wrapped, "_option_", False):
                    continue
                sig = inspect.signature(wrapped)
                if len(sig.parameters) != 1 or next(iter(sig.parameters)) != "self":
                    raise ValueError(
                        f"The @option decorator expects to be placed on a no-arg instance method, but {basecls.__name__}.{name} does not have this signature."
                    )
                option_type = sig.return_annotation
                if not _is_valid_type(option_type):
                    raise ValueError(
                        f"Invalid type `{option_type}` for option {basecls.__name__}.{name}"
                    )

                help = getattr(wrapped, "_option_help_")
                if callable(help):
                    help = help(cls)
                required = getattr(wrapped, "_option_required_")
                default = getattr(wrapped, "_option_default_")
                if required:
                    if default is not None:
                        raise ValueError(
                            "The option {basecls.__name__}.{name} is required, so it must not provide a default value."
                        )
                else:
                    if callable(default):
                        default = default(cls)
                    if default is not None and not _is_of_type(default, option_type):
                        raise ValueError(
                            f"The default for option {basecls.__name__}.{name} must be of type {_type_to_readable_str(option_type)} (or None)."
                        )
                option_descriptors.append(
                    OptionDescriptor(
                        name,
                        type=option_type,
                        required=required,
                        default=default,
                        help=help,
                    )
                )
        cls._option_descriptors_ = tuple(option_descriptors)

    @memoized_classmethod
    def _get_rules_(cls: Any) -> Iterable[Rule]:
        return cast(Iterable[Rule], (cls._construct_subsystem_rule_(),))

    @classmethod
    def _get_construct_func_(cls) -> Tuple[Callable, dict]:
        """Returns information that allows the engine to construct an instance of the subsystem.

        The return value is a pair (function, params) where the function takes the
        type of the subsystem as its first argument (`cls`) and the given params
        as its subsequent argument(s). The params dict is a map of param name -> param type.

        See below for examples of use (currently the only uses, in fact).
        """
        raise ValueError(f"_get_construct_func_() not implemented for {cls}")

    @classmethod
    def _construct_subsystem_rule_(cls) -> Rule:
        """Returns a `TaskRule` that will construct the target Subsystem."""
        construct_func, construct_func_params = cls._get_construct_func_()

        partial_construct_subsystem: Any = functools.partial(construct_func, cls)

        # NB: We populate several dunder methods on the partial function because partial
        # functions do not have these defined by default, and the engine uses these values to
        # visualize functions in error messages and the rule graph.
        name = f"construct_{cls.__name__}"
        partial_construct_subsystem.__name__ = name
        partial_construct_subsystem.__module__ = cls.__module__
        partial_construct_subsystem.__doc__ = cls.help() if callable(cls.help) else cls.help
        partial_construct_subsystem.__line_number__ = inspect.getsourcelines(cls)[1]

        return TaskRule(
            output_type=cls,
            parameters=FrozenDict(construct_func_params),
            awaitables=(),
            masked_types=(),
            func=partial_construct_subsystem,
            canonical_name=name,
        )


@dataclass(frozen=True)
class UniversalOptionsReader:
    options_reader: PyNgOptionsReader


@rule
async def get_universal_options_reader(options: PyNgOptions) -> UniversalOptionsReader:
    # The universal options are the options for the repo root dir.
    return UniversalOptionsReader(options.get_options_reader_for_dir(""))


class UniversalSubsystem(SubsystemNg):
    """A subsystem whose values are universal for all uses.

    Uses include:
    - Global options, as these affect the running of Pants itself.
    - Command/subcommand options, as these are typically provided on the command line and
      are intended by the user to apply to all sources.
    """

    @classmethod
    def _get_construct_func_(cls):
        return _construct_universal_subsystem, {"universal_options_reader": UniversalOptionsReader}


_UniversalSubsystemT = TypeVar("_UniversalSubsystemT", bound="UniversalSubsystem")


async def _construct_universal_subsystem(
    subsystem_typ: type[_UniversalSubsystemT],
    universal_options_reader: UniversalOptionsReader,
) -> _UniversalSubsystemT:
    return subsystem_typ(universal_options_reader.options_reader)


class ContextualSubsystem(SubsystemNg):
    """A subsystem whose values may vary depending on the sources it applies to.

    The main use is to allow config files in subdirectories to override option values
    set in parent directories (including the root directory), and for those overrides
    to apply just to the sources in or below those subdirectories.

    The input sources for a command will be partitioned into subsets, each attached to
    its own subsystem instance.
    """

    @classmethod
    def _get_construct_func_(cls):
        return _construct_contextual_subsystem, {"source_partition": SourcePartition}


_ContextualSubsystemT = TypeVar("_ContextualSubsystemT", bound="ContextualSubsystem")


async def _construct_contextual_subsystem(
    subsystem_typ: type[_ContextualSubsystemT],
    source_partition: SourcePartition,
) -> _ContextualSubsystemT:
    return subsystem_typ(source_partition.options_reader)


def rules() -> tuple[Rule, ...]:
    return tuple(collect_rules())

1	# Copyright 2025 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	import functools	1✔
5	import inspect	1✔
6	import typing	1✔
7	from collections.abc import Callable	1✔
8	from dataclasses import dataclass	1✔
9	from functools import partial, wraps	1✔
10	from types import GenericAlias	1✔
11	from typing import Any, Iterable, Self, Tuple, TypeVar, cast	1✔
12
13	from pants.engine.internals.native_engine import PyNgOptions, PyNgOptionsReader, PyOptionId	1✔
14	from pants.engine.rules import Rule, TaskRule, collect_rules, rule	1✔
15	from pants.ng.source_partition import SourcePartition	1✔
16	from pants.option.ranked_value import Rank	1✔
17	from pants.util.frozendict import FrozenDict	1✔
18	from pants.util.memo import memoized_classmethod	1✔
19
20	# Supported option value types.
21	#
22	# Note that this is a subset of the value types supported by og Pants, which allows
23	# nested and heterogenously-valued dicts. However that doesn't seem to be worth the effort
24	# so let's see how far we can get without supporting those. Note also that these types are all
25	# immutable, whereas the og types did not have to be. Again, not a complication we should
26	# be enthusiastic to support.
27	ScalarValue = bool \| str \| int \| float	1✔
28	OptionValue = ScalarValue \| tuple[ScalarValue, ...] \| FrozenDict[str, str]	1✔
29
30
31	_SCALAR_TYPES = typing.get_args(ScalarValue)	1✔
32
33
34	def _is_valid_type(t: type \| GenericAlias) -> bool:	1✔
35	if isinstance(t, type):	1✔
36	# ScalarType
37	return t in _SCALAR_TYPES	1✔
38
39	torig = typing.get_origin(t)	1✔
40	targs = typing.get_args(t)	1✔
41	return (	1✔
42	# tuple[ScalarType, ...]
43	torig == tuple and len(targs) == 2 and targs[0] in _SCALAR_TYPES and targs[1] == Ellipsis
44	) or (
45	# FrozenDict[str, str]
46	torig == FrozenDict and targs == (str, str)
47	)
48
49
50	def _type_to_readable_str(t: type \| GenericAlias) -> str:	1✔
51	if isinstance(t, type):	1✔
52	# E.g., `str` instead of `<class 'str'>`
53	return t.__name__	1✔
54	# A GenericAlias renders as you'd expect (e.g., `tuple[str, ...]`).
55	return str(t)	1✔
56
57
58	def _is_of_type(val: OptionValue, expected_type: type \| GenericAlias) -> bool:	1✔
59	if isinstance(expected_type, type) and expected_type in _SCALAR_TYPES:	1✔
60	return isinstance(val, expected_type)	1✔
61
62	torig = typing.get_origin(expected_type)	1✔
63	targs = typing.get_args(expected_type)	1✔
64	if torig == tuple:	1✔
65	member_type = targs[0]	1✔
66	return isinstance(val, tuple) and all(isinstance(member, member_type) for member in val)	1✔
67	if torig == FrozenDict:	1✔
68	return isinstance(val, FrozenDict) and all(	1✔
69	(type(k), type(v)) == (str, str) for k, v in val.items()
70	)
71	# Should never happen, assuming we've already passed _is_valid_type().
NEW 72	raise ValueError(f"Unexpected expected_type {expected_type}")	×
73
74
75	def _getter(getter: Callable, option_parser, option_id, default) -> tuple[Any, bool]:	1✔
76	# The legacy getter returns a tuple of (value, rank, optional derivation),
77	# we only care about value and rank.
78	value, rank, _ = getter(option_parser, option_id, default)	1✔
79	return value, rank > Rank.HARDCODED._rank	1✔
80
81
82	def _tuple_getter(	1✔
83	getter: Callable, option_parser, option_id, default
84	) -> tuple[tuple[ScalarValue, ...], bool]:
85	# The legacy getter returns a tuple of (list value, rank, optional derivation),
86	# we only care about value and rank, and we convert the value to tuple. The rust code can
87	# consume a tuple where a list is expected, so no need to convert the default.
88	value, rank, _ = getter(option_parser, option_id, default or ())	1✔
89	return tuple(value), rank > Rank.HARDCODED._rank	1✔
90
91
92	def _dict_getter(option_parser, option_id, default) -> tuple[FrozenDict, bool]:	1✔
93	# The legacy getter returns a tuple of (value, rank, optional derivation),
94	# we only care about value and rank, and we convert the value to FrozenDict.
95	# We also convert the default to regular dict, so the Rust code can consume it.
96	# The type checker currently doesn't grok the overloaded FrozenDict.__init__, so
97	# we ignore the arg-type error.
98	value, rank, _ = PyNgOptionsReader.get_dict(option_parser, option_id, dict(default or {}))	1✔
99	return FrozenDict(value), rank > Rank.HARDCODED._rank # type: ignore[arg-type]	1✔
100
101
102	# A map from type to getter for that type.
103	# Each getter is called with the arguments (options_reader, option_id, default) and
104	# returns a pair (value, is_explicit). is_explicit is True if the value was specified
105	# explicitly (by config, env var or flag), and False if the default was used.
106	# Note that this differentiates between the case of "no explicit value" vs "explicit value
107	# was provided, but it happened to be the same as the default value".
108	_getters: dict[type \| GenericAlias, Callable] = {	1✔
109	bool: partial(_getter, PyNgOptionsReader.get_bool),
110	str: partial(_getter, PyNgOptionsReader.get_string),
111	int: partial(_getter, PyNgOptionsReader.get_int),
112	float: partial(_getter, PyNgOptionsReader.get_float),
113	tuple[bool, ...]: partial(_tuple_getter, PyNgOptionsReader.get_bool_list),
114	tuple[str, ...]: partial(_tuple_getter, PyNgOptionsReader.get_string_list),
115	tuple[int, ...]: partial(_tuple_getter, PyNgOptionsReader.get_int_list),
116	tuple[float, ...]: partial(_tuple_getter, PyNgOptionsReader.get_float_list),
117	FrozenDict[str, str]: _dict_getter,
118	}
119
120
121	@dataclass(frozen=True)	1✔
122	class OptionDescriptor:	1✔
123	"""Information we capture at @option evaluation time."""
124
125	name: str
126	type: type \| GenericAlias
127	default: OptionValue \| None
128	help: str
129	required: bool = False	1✔
130
131
132	R = TypeVar("R")	1✔
133	S = TypeVar("S")	1✔
134	OptionFunc = Callable[[S], R]	1✔
135	Default = OptionValue \| Callable[[Any], OptionValue]	1✔
136	Help = str \| Callable[[Any], str]	1✔
137
138
139	REQUIRED = object()	1✔
140
141
142	# A @decorator for options.
143	def option(*, help: Help, required: bool = False, default: Default \| None = None) -> OptionFunc:	1✔
144	def decorator(func: OptionFunc):	1✔
145	# Mark the func as an option.
146	func._option_ = True # type: ignore[attr-defined]	1✔
147	func._option_required_ = required # type: ignore[attr-defined]	1✔
148	func._option_default_ = default # type: ignore[attr-defined]	1✔
149	func._option_help_ = help # type: ignore[attr-defined]	1✔
150
151	@wraps(func)	1✔
152	def wrapper(instance):	1✔
153	return instance._option_values_[func.__name__]	1✔
154
155	return property(wrapper)	1✔
156
157	return decorator	1✔
158
159
160	class SubsystemNg:	1✔
161	"""A holder of options in a scope.
162
163	Much lighter weight than its og counterpart.
164
165	Use like this:
166
167	class Foo(SubsystemNg):
168	options_scope = "foo"
169
170	@option(help="bar help")
171	def bar(self) -> str: ...
172
173	@option(default=42, help="baz help")
174	def baz(self) -> int: ...
175
176	...
177
178	The supported option types are: bool, str, int, float, tuple[bool, ...], tuple[str, ...],
179	tuple[int, ...], tuple[float, ...] and FrozenDict[str, str].
180
181	The `...` in the method bodies above are literal: the subsystem mechanism will provide the body.
182	You may need `# mypy: disable-error-code=empty-body` for your code to pass mypy type checks, or
183	you can use the mypy plugin at pants.ng.subsystem_mypy_plugin to suppress those errors.
184
185	`default` and `help` can be values or callables that take the subsystem type as an arg, and
186	return the relevant values.
187
188	Subsystem instances are instantiated with a PyNgOptionsReader, and calling .bar or .baz (with
189	no parentheses) on an instance of Foo will return the value of that option from the underlying
190	parser.
191
192	Option names must not _begin_and_end_ with an underscore. Such names are reserved for the
193	underlying initialization mechanics.
194	"""
195
196	_subsystem_ng_ = True	1✔
197
198	# Subclasses must set.
199	options_scope: str \| None = None	1✔
200	help: str \| None = None	1✔
201
202	# The ctor will set.
203	_option_values_: FrozenDict[str, OptionValue] \| None = None	1✔
204
205	@classmethod	1✔
206	def create(cls, options_reader: PyNgOptionsReader) -> Self:	1✔
207	"""Indirect constructor method.
208
209	Useful for avoiding spurious mypy errors when instantiating instances in tests.
210	"""
211	return cls(options_reader)	1✔
212
213	def __init__(self, options_reader: PyNgOptionsReader):	1✔
214	option_values = {}	1✔
215	for descriptor in self._get_option_descriptors_():	1✔
216	option_id = PyOptionId(descriptor.name, scope=self.options_scope)	1✔
217	getter = _getters[descriptor.type]	1✔
218	val, is_explicit = getter(	1✔
219	options_reader, option_id=option_id, default=descriptor.default
220	)
221	if descriptor.required and not is_explicit:	1✔
222	raise ValueError(	1✔
223	"No value provided for required option "
224	f"[{self.options_scope}].{descriptor.name}."
225	)
226	option_values[descriptor.name] = val	1✔
227	self._option_values_ = FrozenDict(option_values)	1✔
228
229	def _get_option_values_(self) -> FrozenDict[str, OptionValue]:	1✔
NEW 230	return cast(FrozenDict[str, OptionValue], self._option_values_)	×
231
232	def __str__(self) -> str:	1✔
NEW 233	fields = [f"{k}={v}" for k, v in self._get_option_values_().items()]	×
NEW 234	return f"{self.__class__.__name__}({', '.join(fields)})"	×
235
236	@classmethod	1✔
237	def _get_option_descriptors_(cls) -> tuple[OptionDescriptor, ...]:	1✔
238	"""Return the option descriptors for this subsystem.
239
240	Must only be called after initialize() has been called.
241	"""
242	return cast(tuple[OptionDescriptor], getattr(cls, "_option_descriptors_"))	1✔
243
244	@classmethod	1✔
245	def _initialize_(cls):	1✔
246	if getattr(cls, "_option_descriptors_", None) is not None:	1✔
NEW 247	return	×
248
249	if cls.options_scope is None:	1✔
250	raise ValueError(f"Subsystem class {cls.__name__} must set the options_scope classvar.")	1✔
251	if cls.help is None:	1✔
252	raise ValueError(f"Subsystem class {cls.__name__} must set the help classvar.")	1✔
253	option_descriptors = []	1✔
254	# We don't use dir() or inspect.getmembers() because those sort by name,
255	# and we want to preserve declaration order.
256	for basecls in inspect.getmro(cls):	1✔
257	for name, member in basecls.__dict__.items():	1✔
258	if not isinstance(member, property):	1✔
259	continue	1✔
260	wrapped = member.fget	1✔
261	if not getattr(wrapped, "_option_", False):	1✔
NEW 262	continue	×
263	sig = inspect.signature(wrapped)	1✔
264	if len(sig.parameters) != 1 or next(iter(sig.parameters)) != "self":	1✔
265	raise ValueError(	1✔
266	f"The @option decorator expects to be placed on a no-arg instance method, but {basecls.__name__}.{name} does not have this signature."
267	)
268	option_type = sig.return_annotation	1✔
269	if not _is_valid_type(option_type):	1✔
NEW 270	raise ValueError(	×
271	f"Invalid type `{option_type}` for option {basecls.__name__}.{name}"
272	)
273
274	help = getattr(wrapped, "_option_help_")	1✔
275	if callable(help):	1✔
276	help = help(cls)	1✔
277	required = getattr(wrapped, "_option_required_")	1✔
278	default = getattr(wrapped, "_option_default_")	1✔
279	if required:	1✔
280	if default is not None:	1✔
NEW 281	raise ValueError(	×
282	"The option {basecls.__name__}.{name} is required, so it must not provide a default value."
283	)
284	else:
285	if callable(default):	1✔
286	default = default(cls)	1✔
287	if default is not None and not _is_of_type(default, option_type):	1✔
288	raise ValueError(	1✔
289	f"The default for option {basecls.__name__}.{name} must be of type {_type_to_readable_str(option_type)} (or None)."
290	)
291	option_descriptors.append(	1✔
292	OptionDescriptor(
293	name,
294	type=option_type,
295	required=required,
296	default=default,
297	help=help,
298	)
299	)
300	cls._option_descriptors_ = tuple(option_descriptors)	1✔
301
302	@memoized_classmethod	1✔
303	def _get_rules_(cls: Any) -> Iterable[Rule]:	1✔
NEW 304	return cast(Iterable[Rule], (cls._construct_subsystem_rule_(),))	×
305
306	@classmethod	1✔
307	def _get_construct_func_(cls) -> Tuple[Callable, dict]:	1✔
308	"""Returns information that allows the engine to construct an instance of the subsystem.
309
310	The return value is a pair (function, params) where the function takes the
311	type of the subsystem as its first argument (`cls`) and the given params
312	as its subsequent argument(s). The params dict is a map of param name -> param type.
313
314	See below for examples of use (currently the only uses, in fact).
315	"""
NEW 316	raise ValueError(f"_get_construct_func_() not implemented for {cls}")	×
317
318	@classmethod	1✔
319	def _construct_subsystem_rule_(cls) -> Rule:	1✔
320	"""Returns a `TaskRule` that will construct the target Subsystem."""
NEW 321	construct_func, construct_func_params = cls._get_construct_func_()	×
322
NEW 323	partial_construct_subsystem: Any = functools.partial(construct_func, cls)	×
324
325	# NB: We populate several dunder methods on the partial function because partial
326	# functions do not have these defined by default, and the engine uses these values to
327	# visualize functions in error messages and the rule graph.
NEW 328	name = f"construct_{cls.__name__}"	×
NEW 329	partial_construct_subsystem.__name__ = name	×
NEW 330	partial_construct_subsystem.__module__ = cls.__module__	×
NEW 331	partial_construct_subsystem.__doc__ = cls.help() if callable(cls.help) else cls.help	×
NEW 332	partial_construct_subsystem.__line_number__ = inspect.getsourcelines(cls)[1]	×
333
NEW 334	return TaskRule(	×
335	output_type=cls,
336	parameters=FrozenDict(construct_func_params),
337	awaitables=(),
338	masked_types=(),
339	func=partial_construct_subsystem,
340	canonical_name=name,
341	)
342
343
344	@dataclass(frozen=True)	1✔
345	class UniversalOptionsReader:	1✔
346	options_reader: PyNgOptionsReader
347
348
349	@rule	1✔
350	async def get_universal_options_reader(options: PyNgOptions) -> UniversalOptionsReader:	1✔
351	# The universal options are the options for the repo root dir.
NEW 352	return UniversalOptionsReader(options.get_options_reader_for_dir(""))	×
353
354
355	class UniversalSubsystem(SubsystemNg):	1✔
356	"""A subsystem whose values are universal for all uses.
357
358	Uses include:
359	- Global options, as these affect the running of Pants itself.
360	- Command/subcommand options, as these are typically provided on the command line and
361	are intended by the user to apply to all sources.
362	"""
363
364	@classmethod	1✔
365	def _get_construct_func_(cls):	1✔
NEW 366	return _construct_universal_subsystem, {"universal_options_reader": UniversalOptionsReader}	×
367
368
369	_UniversalSubsystemT = TypeVar("_UniversalSubsystemT", bound="UniversalSubsystem")	1✔
370
371
372	async def _construct_universal_subsystem(	1✔
373	subsystem_typ: type[_UniversalSubsystemT],
374	universal_options_reader: UniversalOptionsReader,
375	) -> _UniversalSubsystemT:
NEW 376	return subsystem_typ(universal_options_reader.options_reader)	×
377
378
379	class ContextualSubsystem(SubsystemNg):	1✔
380	"""A subsystem whose values may vary depending on the sources it applies to.
381
382	The main use is to allow config files in subdirectories to override option values
383	set in parent directories (including the root directory), and for those overrides
384	to apply just to the sources in or below those subdirectories.
385
386	The input sources for a command will be partitioned into subsets, each attached to
387	its own subsystem instance.
388	"""
389
390	@classmethod	1✔
391	def _get_construct_func_(cls):	1✔
NEW 392	return _construct_contextual_subsystem, {"source_partition": SourcePartition}	×
393
394
395	_ContextualSubsystemT = TypeVar("_ContextualSubsystemT", bound="ContextualSubsystem")	1✔
396
397
398	async def _construct_contextual_subsystem(	1✔
399	subsystem_typ: type[_ContextualSubsystemT],
400	source_partition: SourcePartition,
401	) -> _ContextualSubsystemT:
NEW 402	return subsystem_typ(source_partition.options_reader)	×
403
404
405	def rules() -> tuple[Rule, ...]:	1✔
NEW 406	return tuple(collect_rules())	×

pantsbuild / pants / 23274822986

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