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/backend/nfpm/util_rules/inject_config.py

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

from __future__ import annotations

from abc import ABC, ABCMeta, abstractmethod
from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any, ClassVar, TypeVar, cast

from pants.backend.nfpm.fields.scripts import NfpmPackageScriptsField
from pants.engine.environment import EnvironmentName
from pants.engine.internals.native_engine import Address, Field
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import Target
from pants.engine.unions import UnionMembership, union
from pants.util.frozendict import FrozenDict
from pants.util.strutil import softwrap

# NB: This TypeVar serves the same purpose here as in pants.engine.target
_F = TypeVar("_F", bound=Field)


@dataclass(frozen=True)
class InjectedNfpmPackageFields:
    """The injected fields that should be used instead of the target's fields.

    Though any field can technically be provided (except "scripts" which is banned),
    only nfpm package metadata fields will have an impact. Passing other fields are
    silently ignored. For example, "dependencies", and "output_path" are not used
    when generating nfpm config, so they will be ignored; "sources" is not a valid
    field for nfpm package targets, so it will also be ignored.

    The "scripts" field is special in that it has dependency inference tied to it.
    If you write your own dependency inference rule (possibly based on a custom
    field you've added to the nfpm package target), then you can pass
    _allow_banned_fields=True to allow injection of the "scripts" field.
    """

    field_values: FrozenDict[type[Field], Field]

    def __init__(
        self,
        fields: Iterable[Field],
        *,
        address: Address,
        _allow_banned_fields: bool = False,
    ) -> None:
        super().__init__()
        if not _allow_banned_fields:
            aliases = [field.alias for field in fields]
            for alias in {
                NfpmPackageScriptsField.alias,  # if _allow_banned_fields, the plugin author must handle scripts deps.
            }:
                if alias in aliases:
                    raise ValueError(
                        softwrap(
                            f"""
                            {alias} cannot be an injected nfpm package field for {address} to avoid
                            breaking dependency inference.
                            """
                        )
                    )
        # Ignore any fields that do not have a value (assuming nfpm fields have 'none_is_valid_value=False').
        field_values = {type(field): field for field in fields if field.value is not None}
        object.__setattr__(
            self,
            "field_values",
            FrozenDict(
                sorted(
                    field_values.items(),
                    key=lambda field_type_to_val_pair: field_type_to_val_pair[0].alias,
                )
            ),
        )


class _PrioritizedSortableClassMetaclass(ABCMeta):
    """This metaclass implements prioritized sorting of subclasses (not class instances)."""

    priority: ClassVar[int]

    def __lt__(self, other: Any) -> bool:
        """Determine if this class is lower priority than `other` (when chaining request rules).

        The rule that runs the lowest priority request goes first, and the rule that runs the
        highest priority request goes last. The results (the `injected_fields`) of lower priority
        rules can be overridden by higher priority rules. The last rule to run, the rule for the
        highest priority request class, can override any of the fields injected by lower priority
        request rules.
        """
        if not isinstance(other, _PrioritizedSortableClassMetaclass):
            return NotImplemented
        if self.priority != other.priority:
            return self.priority < other.priority
        # other has same priority: fall back to name comparison (ensures deterministic sort)
        return (self.__module__, self.__qualname__) < (other.__module__, other.__qualname__)


# Note: This only exists as a hook for additional logic for nFPM config generation, e.g. for plugin
# authors. To resolve `InjectedNfpmPackageFields`, call `determine_injected_nfpm_package_fields`,
# which handles running any custom implementations vs. using the default implementation.
@union(in_scope_types=[EnvironmentName])
@dataclass(frozen=True)
class InjectNfpmPackageFieldsRequest(ABC, metaclass=_PrioritizedSortableClassMetaclass):
    """A request to inject nFPM config for nfpm_package_* targets.

    By default, Pants will use the nfpm_package_* fields in the BUILD file unchanged to generate the
    nfpm.yaml config file for nFPM. To customize this, subclass `InjectNfpmPackageFieldsRequest`,
    register `UnionRule(InjectNfpmPackageFieldsRequest, MyCustomInjectNfpmPackageFieldsRequest)`,
    and add a rule that takes your subclass as a parameter and returns `InjectedNfpmPackageFields`.

    The `target` attribute of this class holds the original target as defined in BUILD files.
    The `injected_fields` attribute of this class contains the results of any previous rule.
    `injected_fields` will be empty for the first rule in the chain. Subsequent rules can remove
    or replace fields injected by previous rules. The final rule in the chain returns the final
    `InjectedNfpmPackageFields` instance that is used to actually generate the nfpm config.
    In general, rules should include a copy of `request.injected_fields` in their return value
    with something like this:

        address = request.target.address
        fields: list[Field] = list(request.injected_fields.values())
        fields.append(NfpmFoobarField("foobar", address))
        return InjectedNfpmPackageFields(fields, address=address)

    Chaining rules like this allows pants to inject some fields, while allowing in-repo plugins
    to override or remove them.
    """

    target: Target
    injected_fields: FrozenDict[type[Field], Field]

    # Classes in pants-provided backends should be priority<10 so that in-repo and external
    # plugins are higher priority by default. This way, the fields that get injected by default
    # can be overridden as needed in the in-repo or external plugins.
    priority: ClassVar[int] = 10

    def __init_subclass__(cls, **kwargs) -> None:
        super().__init_subclass__(**kwargs)
        if cls.priority == getattr(super(), "priority", cls.priority):
            # subclasses are higher priority than their parent class (unless set otherwise).
            cls.priority += 1

    @classmethod
    @abstractmethod
    def is_applicable(cls, target: Target) -> bool:
        """Whether to use this InjectNfpmPackageFieldsRequest implementation for this target."""

    def get_field(self, field: type[_F]) -> _F:
        """Get a `Field` from `injected_fields` (returned by earlier rules) or from the target.

        This will throw a KeyError if the `Field` is not registered on the target (unless an earlier
        rule added it to `injected_fields` which might be disallowed in the future).
        """
        if field in self.injected_fields:
            return cast(_F, self.injected_fields[field])
        return self.target[field]


@rule(polymorphic=True)
async def inject_nfpm_package_fields(
    req: InjectNfpmPackageFieldsRequest, env_name: EnvironmentName
) -> InjectedNfpmPackageFields:
    raise NotImplementedError()


@dataclass(frozen=True)
class NfpmPackageTargetWrapper:
    """Nfpm Package target Wrapper.

    This is not meant to be used by plugin authors.
    """

    target: Target


@rule
async def determine_injected_nfpm_package_fields(
    wrapper: NfpmPackageTargetWrapper, union_membership: UnionMembership
) -> InjectedNfpmPackageFields:
    target = wrapper.target
    inject_nfpm_config_request_types = union_membership.get(InjectNfpmPackageFieldsRequest)

    # Requests are sorted (w/ priority ClassVar) before chaining the rules that take them.
    applicable_inject_nfpm_config_request_types = tuple(
        sorted(
            request_type
            for request_type in inject_nfpm_config_request_types
            if request_type.is_applicable(target)
        )
    )

    # If no provided implementations, fall back to our default implementation that simply returns
    # what the user explicitly specified in the BUILD file.
    if not applicable_inject_nfpm_config_request_types:
        return InjectedNfpmPackageFields((), address=target.address)

    injected: InjectedNfpmPackageFields
    injected_fields: FrozenDict[type[Field], Field] = FrozenDict()
    for request_type in applicable_inject_nfpm_config_request_types:
        chained_request: InjectNfpmPackageFieldsRequest = request_type(target, injected_fields)  # type: ignore[abstract]
        injected = await inject_nfpm_package_fields(
            **implicitly({chained_request: InjectNfpmPackageFieldsRequest})
        )
        injected_fields = injected.field_values

    # return the last result in the chain of results
    return injected


def rules():
    return [
        *collect_rules(),
    ]

1	# Copyright 2025 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	from abc import ABC, ABCMeta, abstractmethod	2✔
7	from collections.abc import Iterable	2✔
8	from dataclasses import dataclass	2✔
9	from typing import Any, ClassVar, TypeVar, cast	2✔
10
11	from pants.backend.nfpm.fields.scripts import NfpmPackageScriptsField	2✔
12	from pants.engine.environment import EnvironmentName	2✔
13	from pants.engine.internals.native_engine import Address, Field	2✔
14	from pants.engine.rules import collect_rules, implicitly, rule	2✔
15	from pants.engine.target import Target	2✔
16	from pants.engine.unions import UnionMembership, union	2✔
17	from pants.util.frozendict import FrozenDict	2✔
18	from pants.util.strutil import softwrap	2✔
19
20	# NB: This TypeVar serves the same purpose here as in pants.engine.target
21	_F = TypeVar("_F", bound=Field)	2✔
22
23
24	@dataclass(frozen=True)	2✔
25	class InjectedNfpmPackageFields:	2✔
26	"""The injected fields that should be used instead of the target's fields.
27
28	Though any field can technically be provided (except "scripts" which is banned),
29	only nfpm package metadata fields will have an impact. Passing other fields are
30	silently ignored. For example, "dependencies", and "output_path" are not used
31	when generating nfpm config, so they will be ignored; "sources" is not a valid
32	field for nfpm package targets, so it will also be ignored.
33
34	The "scripts" field is special in that it has dependency inference tied to it.
35	If you write your own dependency inference rule (possibly based on a custom
36	field you've added to the nfpm package target), then you can pass
37	_allow_banned_fields=True to allow injection of the "scripts" field.
38	"""
39
40	field_values: FrozenDict[type[Field], Field]	2✔
41
42	def __init__(	2✔
43	self,
44	fields: Iterable[Field],
45	*,
46	address: Address,
47	_allow_banned_fields: bool = False,
48	) -> None:
49	super().__init__()	×
50	if not _allow_banned_fields:	×
51	aliases = [field.alias for field in fields]	×
52	for alias in {	×
53	NfpmPackageScriptsField.alias, # if _allow_banned_fields, the plugin author must handle scripts deps.
54	}:
55	if alias in aliases:	×
56	raise ValueError(	×
57	softwrap(
58	f"""
59	{alias} cannot be an injected nfpm package field for {address} to avoid
60	breaking dependency inference.
61	"""
62	)
63	)
64	# Ignore any fields that do not have a value (assuming nfpm fields have 'none_is_valid_value=False').
65	field_values = {type(field): field for field in fields if field.value is not None}	×
66	object.__setattr__(	×
67	self,
68	"field_values",
69	FrozenDict(
70	sorted(
71	field_values.items(),
72	key=lambda field_type_to_val_pair: field_type_to_val_pair[0].alias,
73	)
74	),
75	)
76
77
78	class _PrioritizedSortableClassMetaclass(ABCMeta):	2✔
79	"""This metaclass implements prioritized sorting of subclasses (not class instances)."""
80
81	priority: ClassVar[int]	2✔
82
83	def __lt__(self, other: Any) -> bool:	2✔
84	"""Determine if this class is lower priority than `other` (when chaining request rules).
85
86	The rule that runs the lowest priority request goes first, and the rule that runs the
87	highest priority request goes last. The results (the `injected_fields`) of lower priority
88	rules can be overridden by higher priority rules. The last rule to run, the rule for the
89	highest priority request class, can override any of the fields injected by lower priority
90	request rules.
91	"""
92	if not isinstance(other, _PrioritizedSortableClassMetaclass):	×
93	return NotImplemented	×
94	if self.priority != other.priority:	×
95	return self.priority < other.priority	×
96	# other has same priority: fall back to name comparison (ensures deterministic sort)
97	return (self.__module__, self.__qualname__) < (other.__module__, other.__qualname__)	×
98
99
100	# Note: This only exists as a hook for additional logic for nFPM config generation, e.g. for plugin
101	# authors. To resolve `InjectedNfpmPackageFields`, call `determine_injected_nfpm_package_fields`,
102	# which handles running any custom implementations vs. using the default implementation.
103	@union(in_scope_types=[EnvironmentName])	2✔
104	@dataclass(frozen=True)	2✔
105	class InjectNfpmPackageFieldsRequest(ABC, metaclass=_PrioritizedSortableClassMetaclass):	2✔
106	"""A request to inject nFPM config for nfpm_package_* targets.
107
108	By default, Pants will use the nfpm_package_* fields in the BUILD file unchanged to generate the
109	nfpm.yaml config file for nFPM. To customize this, subclass `InjectNfpmPackageFieldsRequest`,
110	register `UnionRule(InjectNfpmPackageFieldsRequest, MyCustomInjectNfpmPackageFieldsRequest)`,
111	and add a rule that takes your subclass as a parameter and returns `InjectedNfpmPackageFields`.
112
113	The `target` attribute of this class holds the original target as defined in BUILD files.
114	The `injected_fields` attribute of this class contains the results of any previous rule.
115	`injected_fields` will be empty for the first rule in the chain. Subsequent rules can remove
116	or replace fields injected by previous rules. The final rule in the chain returns the final
117	`InjectedNfpmPackageFields` instance that is used to actually generate the nfpm config.
118	In general, rules should include a copy of `request.injected_fields` in their return value
119	with something like this:
120
121	address = request.target.address
122	fields: list[Field] = list(request.injected_fields.values())
123	fields.append(NfpmFoobarField("foobar", address))
124	return InjectedNfpmPackageFields(fields, address=address)
125
126	Chaining rules like this allows pants to inject some fields, while allowing in-repo plugins
127	to override or remove them.
128	"""
129
130	target: Target	2✔
131	injected_fields: FrozenDict[type[Field], Field]	2✔
132
133	# Classes in pants-provided backends should be priority<10 so that in-repo and external
134	# plugins are higher priority by default. This way, the fields that get injected by default
135	# can be overridden as needed in the in-repo or external plugins.
136	priority: ClassVar[int] = 10	2✔
137
138	def __init_subclass__(cls, **kwargs) -> None:	2✔
139	super().__init_subclass__(**kwargs)	×
140	if cls.priority == getattr(super(), "priority", cls.priority):	×
141	# subclasses are higher priority than their parent class (unless set otherwise).
142	cls.priority += 1	×
143
144	@classmethod	2✔
145	@abstractmethod	2✔
146	def is_applicable(cls, target: Target) -> bool:	2✔
147	"""Whether to use this InjectNfpmPackageFieldsRequest implementation for this target."""
148
149	def get_field(self, field: type[_F]) -> _F:	2✔
150	"""Get a `Field` from `injected_fields` (returned by earlier rules) or from the target.
151
152	This will throw a KeyError if the `Field` is not registered on the target (unless an earlier
153	rule added it to `injected_fields` which might be disallowed in the future).
154	"""
155	if field in self.injected_fields:	×
156	return cast(_F, self.injected_fields[field])	×
157	return self.target[field]	×
158
159
160	@rule(polymorphic=True)	2✔
161	async def inject_nfpm_package_fields(	2✔
162	req: InjectNfpmPackageFieldsRequest, env_name: EnvironmentName
163	) -> InjectedNfpmPackageFields:
164	raise NotImplementedError()	×
165
166
167	@dataclass(frozen=True)	2✔
168	class NfpmPackageTargetWrapper:	2✔
169	"""Nfpm Package target Wrapper.
170
171	This is not meant to be used by plugin authors.
172	"""
173
174	target: Target	2✔
175
176
177	@rule	2✔
178	async def determine_injected_nfpm_package_fields(	2✔
179	wrapper: NfpmPackageTargetWrapper, union_membership: UnionMembership
180	) -> InjectedNfpmPackageFields:
181	target = wrapper.target	×
182	inject_nfpm_config_request_types = union_membership.get(InjectNfpmPackageFieldsRequest)	×
183
184	# Requests are sorted (w/ priority ClassVar) before chaining the rules that take them.
185	applicable_inject_nfpm_config_request_types = tuple(	×
186	sorted(
187	request_type
188	for request_type in inject_nfpm_config_request_types
189	if request_type.is_applicable(target)
190	)
191	)
192
193	# If no provided implementations, fall back to our default implementation that simply returns
194	# what the user explicitly specified in the BUILD file.
195	if not applicable_inject_nfpm_config_request_types:	×
196	return InjectedNfpmPackageFields((), address=target.address)	×
197
198	injected: InjectedNfpmPackageFields
199	injected_fields: FrozenDict[type[Field], Field] = FrozenDict()	×
200	for request_type in applicable_inject_nfpm_config_request_types:	×
201	chained_request: InjectNfpmPackageFieldsRequest = request_type(target, injected_fields) # type: ignore[abstract]	×
202	injected = await inject_nfpm_package_fields(	×
203	**implicitly({chained_request: InjectNfpmPackageFieldsRequest})
204	)
205	injected_fields = injected.field_values	×
206
207	# return the last result in the chain of results
208	return injected	×
209
210
211	def rules():	2✔
212	return [	2✔
213	*collect_rules(),
214	]

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