• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pantsbuild / pants / 24055979590

06 Apr 2026 11:17PM UTC coverage: 52.37% (-40.5%) from 92.908%
24055979590

Pull #23225

github

web-flow
Merge 67474653c into 542ca048d
Pull Request #23225: Add --test-show-all-batch-targets to expose all targets in batched pytest

6 of 17 new or added lines in 2 files covered. (35.29%)

23030 existing lines in 605 files now uncovered.

31643 of 60422 relevant lines covered (52.37%)

1.05 hits per line

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

0.0
/src/python/pants/ng/subsystem.py
1
# Copyright 2025 Pants project contributors (see CONTRIBUTORS.md).
2
# Licensed under the Apache License, Version 2.0 (see LICENSE).
3

UNCOV
4
import functools
×
UNCOV
5
import inspect
×
UNCOV
6
import typing
×
UNCOV
7
from collections.abc import Callable
×
UNCOV
8
from dataclasses import dataclass
×
UNCOV
9
from functools import partial, wraps
×
UNCOV
10
from types import GenericAlias
×
UNCOV
11
from typing import Any, Iterable, Self, Tuple, TypeVar, cast
×
12

UNCOV
13
from pants.engine.internals.native_engine import PyNgOptions, PyNgOptionsReader, PyOptionId
×
UNCOV
14
from pants.engine.rules import Rule, TaskRule, collect_rules, rule
×
UNCOV
15
from pants.ng.source_partition import SourcePartition
×
UNCOV
16
from pants.option.ranked_value import Rank
×
UNCOV
17
from pants.util.frozendict import FrozenDict
×
UNCOV
18
from pants.util.memo import memoized_classmethod
×
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.
UNCOV
27
ScalarValue = bool | str | int | float
×
UNCOV
28
OptionValue = ScalarValue | tuple[ScalarValue, ...] | FrozenDict[str, str]
×
29

30

UNCOV
31
_SCALAR_TYPES = typing.get_args(ScalarValue)
×
32

33

UNCOV
34
def _is_valid_type(t: type | GenericAlias) -> bool:
×
UNCOV
35
    if isinstance(t, type):
×
36
        # ScalarType
UNCOV
37
        return t in _SCALAR_TYPES
×
38

UNCOV
39
    torig = typing.get_origin(t)
×
UNCOV
40
    targs = typing.get_args(t)
×
UNCOV
41
    return (
×
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

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

57

UNCOV
58
def _is_of_type(val: OptionValue, expected_type: type | GenericAlias) -> bool:
×
UNCOV
59
    if isinstance(expected_type, type) and expected_type in _SCALAR_TYPES:
×
UNCOV
60
        return isinstance(val, expected_type)
×
61

UNCOV
62
    torig = typing.get_origin(expected_type)
×
UNCOV
63
    targs = typing.get_args(expected_type)
×
UNCOV
64
    if torig == tuple:
×
UNCOV
65
        member_type = targs[0]
×
UNCOV
66
        return isinstance(val, tuple) and all(isinstance(member, member_type) for member in val)
×
UNCOV
67
    if torig == FrozenDict:
×
UNCOV
68
        return isinstance(val, FrozenDict) and all(
×
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().
72
    raise ValueError(f"Unexpected expected_type {expected_type}")
×
73

74

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

81

UNCOV
82
def _tuple_getter(
×
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.
UNCOV
88
    value, rank, _ = getter(option_parser, option_id, default or ())
×
UNCOV
89
    return tuple(value), rank > Rank.HARDCODED._rank
×
90

91

UNCOV
92
def _dict_getter(option_parser, option_id, default) -> tuple[FrozenDict, bool]:
×
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.
UNCOV
98
    value, rank, _ = PyNgOptionsReader.get_dict(option_parser, option_id, dict(default or {}))
×
UNCOV
99
    return FrozenDict(value), rank > Rank.HARDCODED._rank  # type: ignore[arg-type]
×
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".
UNCOV
108
_getters: dict[type | GenericAlias, Callable] = {
×
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

UNCOV
121
@dataclass(frozen=True)
×
UNCOV
122
class OptionDescriptor:
×
123
    """Information we capture at @option evaluation time."""
124

125
    name: str
126
    type: type | GenericAlias
127
    default: OptionValue | None
128
    help: str
UNCOV
129
    required: bool = False
×
130

131

UNCOV
132
R = TypeVar("R")
×
UNCOV
133
S = TypeVar("S")
×
UNCOV
134
OptionFunc = Callable[[S], R]
×
UNCOV
135
Default = OptionValue | Callable[[Any], OptionValue]
×
UNCOV
136
Help = str | Callable[[Any], str]
×
137

138

UNCOV
139
REQUIRED = object()
×
140

141

142
# A @decorator for options.
UNCOV
143
def option(*, help: Help, required: bool = False, default: Default | None = None) -> OptionFunc:
×
UNCOV
144
    def decorator(func: OptionFunc):
×
145
        # Mark the func as an option.
UNCOV
146
        func._option_ = True  # type: ignore[attr-defined]
×
UNCOV
147
        func._option_required_ = required  # type: ignore[attr-defined]
×
UNCOV
148
        func._option_default_ = default  # type: ignore[attr-defined]
×
UNCOV
149
        func._option_help_ = help  # type: ignore[attr-defined]
×
150

UNCOV
151
        @wraps(func)
×
UNCOV
152
        def wrapper(instance):
×
UNCOV
153
            return instance._option_values_[func.__name__]
×
154

UNCOV
155
        return property(wrapper)
×
156

UNCOV
157
    return decorator
×
158

159

UNCOV
160
class SubsystemNg:
×
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

UNCOV
196
    _subsystem_ng_ = True
×
197

198
    # Subclasses must set.
UNCOV
199
    options_scope: str | None = None
×
UNCOV
200
    help: str | None = None
×
201

202
    # The ctor will set.
UNCOV
203
    _option_values_: FrozenDict[str, OptionValue] | None = None
×
204

UNCOV
205
    @classmethod
×
UNCOV
206
    def create(cls, options_reader: PyNgOptionsReader) -> Self:
×
207
        """Indirect constructor method.
208

209
        Useful for avoiding spurious mypy errors when instantiating instances in tests.
210
        """
UNCOV
211
        return cls(options_reader)
×
212

UNCOV
213
    def __init__(self, options_reader: PyNgOptionsReader):
×
UNCOV
214
        option_values = {}
×
UNCOV
215
        for descriptor in self._get_option_descriptors_():
×
UNCOV
216
            option_id = PyOptionId(descriptor.name, scope=self.options_scope)
×
UNCOV
217
            getter = _getters[descriptor.type]
×
UNCOV
218
            val, is_explicit = getter(
×
219
                options_reader, option_id=option_id, default=descriptor.default
220
            )
UNCOV
221
            if descriptor.required and not is_explicit:
×
UNCOV
222
                raise ValueError(
×
223
                    "No value provided for required option "
224
                    f"[{self.options_scope}].{descriptor.name}."
225
                )
UNCOV
226
            option_values[descriptor.name] = val
×
UNCOV
227
        self._option_values_ = FrozenDict(option_values)
×
228

UNCOV
229
    def _get_option_values_(self) -> FrozenDict[str, OptionValue]:
×
230
        return cast(FrozenDict[str, OptionValue], self._option_values_)
×
231

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

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

240
        Must only be called after initialize() has been called.
241
        """
UNCOV
242
        return cast(tuple[OptionDescriptor], getattr(cls, "_option_descriptors_"))
×
243

UNCOV
244
    @classmethod
×
UNCOV
245
    def _initialize_(cls):
×
UNCOV
246
        if getattr(cls, "_option_descriptors_", None) is not None:
×
247
            return
×
248

UNCOV
249
        if cls.options_scope is None:
×
UNCOV
250
            raise ValueError(f"Subsystem class {cls.__name__} must set the options_scope classvar.")
×
UNCOV
251
        if cls.help is None:
×
UNCOV
252
            raise ValueError(f"Subsystem class {cls.__name__} must set the help classvar.")
×
UNCOV
253
        option_descriptors = []
×
254
        # We don't use dir() or inspect.getmembers() because those sort by name,
255
        # and we want to preserve declaration order.
UNCOV
256
        for basecls in inspect.getmro(cls):
×
UNCOV
257
            for name, member in basecls.__dict__.items():
×
UNCOV
258
                if not isinstance(member, property):
×
UNCOV
259
                    continue
×
UNCOV
260
                wrapped = member.fget
×
UNCOV
261
                if not getattr(wrapped, "_option_", False):
×
262
                    continue
×
UNCOV
263
                sig = inspect.signature(wrapped)
×
UNCOV
264
                if len(sig.parameters) != 1 or next(iter(sig.parameters)) != "self":
×
UNCOV
265
                    raise ValueError(
×
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
                    )
UNCOV
268
                option_type = sig.return_annotation
×
UNCOV
269
                if not _is_valid_type(option_type):
×
270
                    raise ValueError(
×
271
                        f"Invalid type `{option_type}` for option {basecls.__name__}.{name}"
272
                    )
273

UNCOV
274
                help = getattr(wrapped, "_option_help_")
×
UNCOV
275
                if callable(help):
×
UNCOV
276
                    help = help(cls)
×
UNCOV
277
                required = getattr(wrapped, "_option_required_")
×
UNCOV
278
                default = getattr(wrapped, "_option_default_")
×
UNCOV
279
                if required:
×
UNCOV
280
                    if default is not None:
×
281
                        raise ValueError(
×
282
                            "The option {basecls.__name__}.{name} is required, so it must not provide a default value."
283
                        )
284
                else:
UNCOV
285
                    if callable(default):
×
UNCOV
286
                        default = default(cls)
×
UNCOV
287
                    if default is not None and not _is_of_type(default, option_type):
×
UNCOV
288
                        raise ValueError(
×
289
                            f"The default for option {basecls.__name__}.{name} must be of type {_type_to_readable_str(option_type)} (or None)."
290
                        )
UNCOV
291
                option_descriptors.append(
×
292
                    OptionDescriptor(
293
                        name,
294
                        type=option_type,
295
                        required=required,
296
                        default=default,
297
                        help=help,
298
                    )
299
                )
UNCOV
300
        cls._option_descriptors_ = tuple(option_descriptors)
×
301

UNCOV
302
    @memoized_classmethod
×
UNCOV
303
    def _get_rules_(cls: Any) -> Iterable[Rule]:
×
304
        return cast(Iterable[Rule], (cls._construct_subsystem_rule_(),))
×
305

UNCOV
306
    @classmethod
×
UNCOV
307
    def _get_construct_func_(cls) -> Tuple[Callable, dict]:
×
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
        """
316
        raise ValueError(f"_get_construct_func_() not implemented for {cls}")
×
317

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

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.
328
        name = f"construct_{cls.__name__}"
×
329
        partial_construct_subsystem.__name__ = name
×
330
        partial_construct_subsystem.__module__ = cls.__module__
×
331
        partial_construct_subsystem.__doc__ = cls.help() if callable(cls.help) else cls.help
×
332
        partial_construct_subsystem.__line_number__ = inspect.getsourcelines(cls)[1]
×
333

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

UNCOV
344
@dataclass(frozen=True)
×
UNCOV
345
class UniversalOptionsReader:
×
346
    options_reader: PyNgOptionsReader
347

348

UNCOV
349
@rule
×
UNCOV
350
async def get_universal_options_reader(options: PyNgOptions) -> UniversalOptionsReader:
×
351
    # The universal options are the options for the repo root dir.
352
    return UniversalOptionsReader(options.get_options_reader_for_dir(""))
×
353

354

UNCOV
355
class UniversalSubsystem(SubsystemNg):
×
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

UNCOV
364
    @classmethod
×
UNCOV
365
    def _get_construct_func_(cls):
×
366
        return _construct_universal_subsystem, {"universal_options_reader": UniversalOptionsReader}
×
367

368

UNCOV
369
_UniversalSubsystemT = TypeVar("_UniversalSubsystemT", bound="UniversalSubsystem")
×
370

371

UNCOV
372
async def _construct_universal_subsystem(
×
373
    subsystem_typ: type[_UniversalSubsystemT],
374
    universal_options_reader: UniversalOptionsReader,
375
) -> _UniversalSubsystemT:
376
    return subsystem_typ(universal_options_reader.options_reader)
×
377

378

UNCOV
379
class ContextualSubsystem(SubsystemNg):
×
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

UNCOV
390
    @classmethod
×
UNCOV
391
    def _get_construct_func_(cls):
×
392
        return _construct_contextual_subsystem, {"source_partition": SourcePartition}
×
393

394

UNCOV
395
_ContextualSubsystemT = TypeVar("_ContextualSubsystemT", bound="ContextualSubsystem")
×
396

397

UNCOV
398
async def _construct_contextual_subsystem(
×
399
    subsystem_typ: type[_ContextualSubsystemT],
400
    source_partition: SourcePartition,
401
) -> _ContextualSubsystemT:
402
    return subsystem_typ(source_partition.options_reader)
×
403

404

UNCOV
405
def rules() -> tuple[Rule, ...]:
×
UNCOV
406
    return tuple(collect_rules())
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc