18198316586

Committed 02 Oct 2025 03:50PM UTC coverage: 78.82% (-1.4%) from 80.265%

Build # 18198316586

Build Type

push

github

Committed by

web-flow

Commit Message

Bump serde from 1.0.226 to 1.0.228 in /src/rust (#22723)

Bumps [serde](https://github.com/serde-rs/serde) from 1.0.226 to
1.0.228.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a
href="https://github.com/serde-rs/serde/releases">serde's
releases</a>.</em></p>
<blockquote>
<h2>v1.0.228</h2>
<ul>
<li>Allow building documentation with
<code>RUSTDOCFLAGS='--cfg=docsrs'</code> set for the whole dependency
graph (<a
href="https://redirect.github.com/serde-rs/serde/issues/2995">#2995</a>)</li>
</ul>
<h2>v1.0.227</h2>
<ul>
<li>Documentation improvements (<a
href="https://redirect.github.com/serde-rs/serde/issues/2991">#2991</a>)</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a
href="https://github.com/serde-rs/serde/commit/a866b336f"><code>a866b33</code></a>
Release 1.0.228</li>
<li><a
href="https://github.com/serde-rs/serde/commit/5adc9e816"><code>5adc9e8</code></a>
Merge pull request <a
href="https://redirect.github.com/serde-rs/serde/issues/2995">#2995</a>
from dtolnay/rustdocflags</li>
<li><a
href="https://github.com/serde-rs/serde/commit/ab581789f"><code>ab58178</code></a>
Workaround for RUSTDOCFLAGS='--cfg=docsrs'</li>
<li><a
href="https://github.com/serde-rs/serde/commit/415d9fc56"><code>415d9fc</code></a>
Release 1.0.227</li>
<li><a
href="https://github.com/serde-rs/serde/commit/7c58427e1"><code>7c58427</code></a>
Merge pull request <a
href="https://redirect.github.com/serde-rs/serde/issues/2991">#2991</a>
from dtolnay/inlinecoredoc</li>
<li><a
href="https://github.com/serde-rs/serde/commit/9d3410e3f"><code>9d3410e</code></a>
Merge pull request <a
href="https://redirect.github.com/serde-rs/serde/issues/2992">#2992</a>
from dtolnay/inplaceseed</li>
<li><a
href="https://github.com/serde-rs/serde/commit/2fb6748bf1ff93... (continued)

Run Details

73576 of 93347 relevant lines covered (78.82%)

2.9 hits per line

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

88.24

/src/python/pants/core/util_rules/partitions.py

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

"""Contains the "base" code for plugin APIs which require partitioning."""

from __future__ import annotations

import itertools
from collections.abc import Iterable
from dataclasses import dataclass
from enum import Enum
from typing import Generic, Protocol, TypeVar, overload

from pants.core.goals.multi_tool_goal_helper import SkippableSubsystem
from pants.engine.collection import Collection
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.internals.graph import resolve_source_paths
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import FieldSet, SourcesField, SourcesPathsRequest
from pants.util.memo import memoized
from pants.util.meta import runtime_ignore_subscripts

_FieldSetT = TypeVar("_FieldSetT", bound=FieldSet)


class PartitionerType(Enum):
    """What type of partitioner to use to partition the input specs."""

    CUSTOM = "custom"
    """The plugin author has a rule to go from `RequestType.PartitionRequest` -> `Partitions`."""

    DEFAULT_SINGLE_PARTITION = "default_single_partition"
    """Registers a partitioner which returns the inputs as a single partition

    The returned partition will have no metadata.
    """

    DEFAULT_ONE_PARTITION_PER_INPUT = "default_one_partition_per_input"
    """Registers a partitioner which returns a single-element partition per input.

    Each of the returned partitions will have no metadata.
    """

    def default_rules(self, cls, *, by_file: bool) -> Iterable:
        """Return an iterable of rules defining the default partitioning logic for this
        `PartitionerType`.

        NOTE: Not all `PartitionerType`s have default logic, so this method can return an empty iterable.

        :param by_file: If `True`, rules returned from this method (if any) will compute partitions with
          `str`-type elements, where each `str` value is the path to the resolved source field. If `False`, rule will compute
          partitions with `FieldSet`-type elements.
        """
        if self == PartitionerType.CUSTOM:
            # No default rules.
            return
        elif self == PartitionerType.DEFAULT_SINGLE_PARTITION:
            rules_generator = (
                _single_partition_file_rules if by_file else _single_partition_field_set_rules
            )
            yield from rules_generator(cls)
        elif self == PartitionerType.DEFAULT_ONE_PARTITION_PER_INPUT:
            rules_generator = (
                _partition_per_input_file_rules if by_file else _partition_per_input_field_set_rules
            )
            yield from rules_generator(cls)
        else:
            raise NotImplementedError(f"Partitioner type {self} is missing default rules!")


class PartitionMetadata(Protocol):
    @property
    def description(self) -> str | None: ...


class _EmptyMetadata:
    @property
    def description(self) -> None:
        return None


PartitionMetadataT = TypeVar("PartitionMetadataT", bound=PartitionMetadata)
PartitionElementT = TypeVar("PartitionElementT")


@dataclass(frozen=True)
@runtime_ignore_subscripts
class Partition(Generic[PartitionElementT, PartitionMetadataT]):
    """A collection of 'compatible' inputs for a plugin tool, with optional common metadata.

    Inputs are 'compatible' if it is safe/possible for them to be processed in the same invocation
    of a tool (i.e. two files formatted in the same run of a formatter, or two test targets executed
    in a single test runner process).

    The metadata in a partition (if any) can be any type able to cross a rule boundary, and will be
    provided to the rule which "runs" your tool. If it isn't `None` it must implement the
    `PartitionMetadata` protocol.

    NOTE: Partitions may be further divided into multiple batches before being passed to the tool-running
    rule. When this happens, the same metadata is passed along with each batch.
    """

    elements: tuple[PartitionElementT, ...]
    metadata: PartitionMetadataT


@runtime_ignore_subscripts
class Partitions(Collection[Partition[PartitionElementT, PartitionMetadataT]]):
    """A collection of (<compatible inputs>, <metadata>) pairs.

    When implementing a plugin, one of your rules will return this type, taking in a
    `PartitionRequest` specific to your plugin.

    The return likely will fit into one of:
        - Returning empty partitions: E.g. if your tool is being skipped.
        - Returning one partition. The partition may contain all of the inputs
            (as will likely be the case for target-based plugins) or a subset (which will likely be the
            case for targetless plugins).
        - Returning >1 partition. This might be the case if you can't run
            the tool on all the inputs at once. E.g. having to run a Python tool on XYZ with Py3,
            and files ABC with Py2.
    """

    @overload
    @classmethod
    def single_partition(
        cls, elements: Iterable[PartitionElementT]
    ) -> Partitions[PartitionElementT, _EmptyMetadata]: ...

    @overload
    @classmethod
    def single_partition(
        cls, elements: Iterable[PartitionElementT], *, metadata: PartitionMetadataT
    ) -> Partitions[PartitionElementT, PartitionMetadataT]: ...

    @classmethod
    def single_partition(
        cls, elements: Iterable[PartitionElementT], *, metadata: PartitionMetadataT | None = None
    ) -> Partitions:
        """Helper constructor for implementations that have only one partition."""
        return Partitions([Partition(tuple(elements), metadata or _EmptyMetadata())])


@dataclass(frozen=True)
@runtime_ignore_subscripts
class _BatchBase(Generic[PartitionElementT, PartitionMetadataT], EngineAwareParameter):
    """Base class for a collection of elements that should all be processed together.

    For example, a collection of strings pointing to files that should be linted in one process, or
    a collection of field-sets pointing at tests that should all execute in the same process.
    """

    tool_name: str
    elements: tuple[PartitionElementT, ...]
    partition_metadata: PartitionMetadataT


@dataclass(frozen=True)
@runtime_ignore_subscripts
class _PartitionFieldSetsRequestBase(Generic[_FieldSetT], EngineAwareParameter):
    """Returns a unique type per calling type.

    This serves us 2 purposes:
        1. `<Core Defined Plugin Type>.PartitionRequest` is the unique type used as a union base for plugin registration.
        2. `<Plugin Defined Subclass>.PartitionRequest` is the unique type used as the union member.
    """

    field_sets: tuple[_FieldSetT, ...]


@dataclass(frozen=True)
class _PartitionFilesRequestBase(EngineAwareParameter):
    """Returns a unique type per calling type.

    This serves us 2 purposes:
        1. `<Core Defined Plugin Type>.PartitionRequest` is the unique type used as a union base for plugin registration.
        2. `<Plugin Defined Subclass>.PartitionRequest` is the unique type used as the union member.
    """

    files: tuple[str, ...]


@memoized
def _single_partition_field_set_rules(cls) -> Iterable:
    """Returns a rule that implements a "partitioner" for `PartitionFieldSetsRequest`, which returns
    one partition."""

    @rule(
        canonical_name_suffix=cls.__name__,
        _param_type_overrides={
            "request": cls.PartitionRequest,
            "subsystem": cls.tool_subsystem,
        },
    )
    async def partitioner(
        request: _PartitionFieldSetsRequestBase, subsystem: SkippableSubsystem
    ) -> Partitions[FieldSet, _EmptyMetadata]:
        return Partitions() if subsystem.skip else Partitions.single_partition(request.field_sets)

    return collect_rules(locals())


@memoized
def _single_partition_file_rules(cls) -> Iterable:
    """Returns a rule that implements a "partitioner" for `PartitionFieldSetsRequest`, which returns
    one partition."""

    # NB: This only works if the FieldSet has a single `SourcesField` field. We check here for
    # a better user experience.
    sources_field_name = _get_sources_field_name(cls.field_set_type)

    @rule(
        canonical_name_suffix=cls.__name__,
        _param_type_overrides={
            "request": cls.PartitionRequest,
            "subsystem": cls.tool_subsystem,
        },
    )
    async def partitioner(
        request: _PartitionFieldSetsRequestBase, subsystem: SkippableSubsystem
    ) -> Partitions[str, _EmptyMetadata]:
        assert sources_field_name is not None
        all_sources_paths = await concurrently(
            resolve_source_paths(
                SourcesPathsRequest(getattr(field_set, sources_field_name)), **implicitly()
            )
            for field_set in request.field_sets
        )

        return (
            Partitions()
            if subsystem.skip
            else Partitions.single_partition(
                itertools.chain.from_iterable(
                    sources_paths.files for sources_paths in all_sources_paths
                )
            )
        )

    return collect_rules(locals())


@memoized
def _partition_per_input_field_set_rules(cls) -> Iterable:
    """Returns a rule that implements a "partitioner" for `PartitionFieldSetsRequest`, which returns
    a single-element partition per input."""

    @rule(
        canonical_name_suffix=cls.__name__,
        _param_type_overrides={
            "request": cls.PartitionRequest,
            "subsystem": cls.tool_subsystem,
        },
    )
    async def partitioner(
        request: _PartitionFieldSetsRequestBase, subsystem: SkippableSubsystem
    ) -> Partitions[FieldSet, _EmptyMetadata]:
        return (
            Partitions()
            if subsystem.skip
            else Partitions(
                Partition((field_set,), _EmptyMetadata()) for field_set in request.field_sets
            )
        )

    return collect_rules(locals())


@memoized
def _partition_per_input_file_rules(cls) -> Iterable:
    """Returns a rule that implements a "partitioner" for `PartitionFieldSetsRequest`, which returns
    a single-element partition per input."""

    # NB: This only works if the FieldSet has a single `SourcesField` field. We check here for
    # a better user experience.
    sources_field_name = _get_sources_field_name(cls.field_set_type)

    @rule(
        canonical_name_suffix=cls.__name__,
        _param_type_overrides={
            "request": cls.PartitionRequest,
            "subsystem": cls.tool_subsystem,
        },
    )
    async def partitioner(
        request: _PartitionFieldSetsRequestBase, subsystem: SkippableSubsystem
    ) -> Partitions[str, _EmptyMetadata]:
        assert sources_field_name is not None
        all_sources_paths = await concurrently(
            resolve_source_paths(
                SourcesPathsRequest(getattr(field_set, sources_field_name)), **implicitly()
            )
            for field_set in request.field_sets
        )

        return (
            Partitions()
            if subsystem.skip
            else Partitions(
                Partition((path,), _EmptyMetadata())
                for path in itertools.chain.from_iterable(
                    sources_paths.files for sources_paths in all_sources_paths
                )
            )
        )

    return collect_rules(locals())


def _get_sources_field_name(field_set_type: type[FieldSet]) -> str:
    """Get the name of the one `SourcesField` belonging to the given target type.

    NOTE: The input target type's fields must contain exactly one `SourcesField`.
    Otherwise this method will raise a `TypeError`.
    """

    sources_field_name = None
    for fieldname, fieldtype in field_set_type.fields.items():
        if issubclass(fieldtype, SourcesField):
            if sources_field_name is None:
                sources_field_name = fieldname
                break
            raise TypeError(
                f"Type {field_set_type} has multiple `SourcesField` fields."
                + " Pants can't provide a default partitioner."
            )
    else:
        raise TypeError(
            f"Type {field_set_type} has does not have a `SourcesField` field."
            + " Pants can't provide a default partitioner."
        )
    return sources_field_name

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	"""Contains the "base" code for plugin APIs which require partitioning."""
5
6	from __future__ import annotations	10✔
7
8	import itertools	10✔
9	from collections.abc import Iterable	10✔
10	from dataclasses import dataclass	10✔
11	from enum import Enum	10✔
12	from typing import Generic, Protocol, TypeVar, overload	10✔
13
14	from pants.core.goals.multi_tool_goal_helper import SkippableSubsystem	10✔
15	from pants.engine.collection import Collection	10✔
16	from pants.engine.engine_aware import EngineAwareParameter	10✔
17	from pants.engine.internals.graph import resolve_source_paths	10✔
18	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	10✔
19	from pants.engine.target import FieldSet, SourcesField, SourcesPathsRequest	10✔
20	from pants.util.memo import memoized	10✔
21	from pants.util.meta import runtime_ignore_subscripts	10✔
22
23	_FieldSetT = TypeVar("_FieldSetT", bound=FieldSet)	10✔
24
25
26	class PartitionerType(Enum):	10✔
27	"""What type of partitioner to use to partition the input specs."""
28
29	CUSTOM = "custom"	10✔
30	"""The plugin author has a rule to go from `RequestType.PartitionRequest` -> `Partitions`."""	10✔
31
32	DEFAULT_SINGLE_PARTITION = "default_single_partition"	10✔
33	"""Registers a partitioner which returns the inputs as a single partition	10✔
34
35	The returned partition will have no metadata.
36	"""
37
38	DEFAULT_ONE_PARTITION_PER_INPUT = "default_one_partition_per_input"	10✔
39	"""Registers a partitioner which returns a single-element partition per input.	10✔
40
41	Each of the returned partitions will have no metadata.
42	"""
43
44	def default_rules(self, cls, *, by_file: bool) -> Iterable:	10✔
45	"""Return an iterable of rules defining the default partitioning logic for this
46	`PartitionerType`.
47
48	NOTE: Not all `PartitionerType`s have default logic, so this method can return an empty iterable.
49
50	:param by_file: If `True`, rules returned from this method (if any) will compute partitions with
51	`str`-type elements, where each `str` value is the path to the resolved source field. If `False`, rule will compute
52	partitions with `FieldSet`-type elements.
53	"""
54	if self == PartitionerType.CUSTOM:	10✔
55	# No default rules.
56	return	9✔
57	elif self == PartitionerType.DEFAULT_SINGLE_PARTITION:	10✔
58	rules_generator = (	10✔
59	_single_partition_file_rules if by_file else _single_partition_field_set_rules
60	)
61	yield from rules_generator(cls)	10✔
62	elif self == PartitionerType.DEFAULT_ONE_PARTITION_PER_INPUT:	9✔
63	rules_generator = (	9✔
64	_partition_per_input_file_rules if by_file else _partition_per_input_field_set_rules
65	)
66	yield from rules_generator(cls)	9✔
67	else:
68	raise NotImplementedError(f"Partitioner type {self} is missing default rules!")	×
69
70
71	class PartitionMetadata(Protocol):	10✔
72	@property
73	def description(self) -> str \| None: ...
74
75
76	class _EmptyMetadata:	10✔
77	@property	10✔
78	def description(self) -> None:	10✔
79	return None	×
80
81
82	PartitionMetadataT = TypeVar("PartitionMetadataT", bound=PartitionMetadata)	10✔
83	PartitionElementT = TypeVar("PartitionElementT")	10✔
84
85
86	@dataclass(frozen=True)	10✔
87	@runtime_ignore_subscripts	10✔
88	class Partition(Generic[PartitionElementT, PartitionMetadataT]):	10✔
89	"""A collection of 'compatible' inputs for a plugin tool, with optional common metadata.
90
91	Inputs are 'compatible' if it is safe/possible for them to be processed in the same invocation
92	of a tool (i.e. two files formatted in the same run of a formatter, or two test targets executed
93	in a single test runner process).
94
95	The metadata in a partition (if any) can be any type able to cross a rule boundary, and will be
96	provided to the rule which "runs" your tool. If it isn't `None` it must implement the
97	`PartitionMetadata` protocol.
98
99	NOTE: Partitions may be further divided into multiple batches before being passed to the tool-running
100	rule. When this happens, the same metadata is passed along with each batch.
101	"""
102
103	elements: tuple[PartitionElementT, ...]	10✔
104	metadata: PartitionMetadataT	10✔
105
106
107	@runtime_ignore_subscripts	10✔
108	class Partitions(Collection[Partition[PartitionElementT, PartitionMetadataT]]):	10✔
109	"""A collection of (<compatible inputs>, <metadata>) pairs.
110
111	When implementing a plugin, one of your rules will return this type, taking in a
112	`PartitionRequest` specific to your plugin.
113
114	The return likely will fit into one of:
115	- Returning empty partitions: E.g. if your tool is being skipped.
116	- Returning one partition. The partition may contain all of the inputs
117	(as will likely be the case for target-based plugins) or a subset (which will likely be the
118	case for targetless plugins).
119	- Returning >1 partition. This might be the case if you can't run
120	the tool on all the inputs at once. E.g. having to run a Python tool on XYZ with Py3,
121	and files ABC with Py2.
122	"""
123
124	@overload
125	@classmethod
126	def single_partition(
127	cls, elements: Iterable[PartitionElementT]
128	) -> Partitions[PartitionElementT, _EmptyMetadata]: ...
129
130	@overload
131	@classmethod
132	def single_partition(
133	cls, elements: Iterable[PartitionElementT], *, metadata: PartitionMetadataT
134	) -> Partitions[PartitionElementT, PartitionMetadataT]: ...
135
136	@classmethod	10✔
137	def single_partition(	10✔
138	cls, elements: Iterable[PartitionElementT], *, metadata: PartitionMetadataT \| None = None
139	) -> Partitions:
140	"""Helper constructor for implementations that have only one partition."""
141	return Partitions([Partition(tuple(elements), metadata or _EmptyMetadata())])	1✔
142
143
144	@dataclass(frozen=True)	10✔
145	@runtime_ignore_subscripts	10✔
146	class _BatchBase(Generic[PartitionElementT, PartitionMetadataT], EngineAwareParameter):	10✔
147	"""Base class for a collection of elements that should all be processed together.
148
149	For example, a collection of strings pointing to files that should be linted in one process, or
150	a collection of field-sets pointing at tests that should all execute in the same process.
151	"""
152
153	tool_name: str	10✔
154	elements: tuple[PartitionElementT, ...]	10✔
155	partition_metadata: PartitionMetadataT	10✔
156
157
158	@dataclass(frozen=True)	10✔
159	@runtime_ignore_subscripts	10✔
160	class _PartitionFieldSetsRequestBase(Generic[_FieldSetT], EngineAwareParameter):	10✔
161	"""Returns a unique type per calling type.
162
163	This serves us 2 purposes:
164	1. `<Core Defined Plugin Type>.PartitionRequest` is the unique type used as a union base for plugin registration.
165	2. `<Plugin Defined Subclass>.PartitionRequest` is the unique type used as the union member.
166	"""
167
168	field_sets: tuple[_FieldSetT, ...]	10✔
169
170
171	@dataclass(frozen=True)	10✔
172	class _PartitionFilesRequestBase(EngineAwareParameter):	10✔
173	"""Returns a unique type per calling type.
174
175	This serves us 2 purposes:
176	1. `<Core Defined Plugin Type>.PartitionRequest` is the unique type used as a union base for plugin registration.
177	2. `<Plugin Defined Subclass>.PartitionRequest` is the unique type used as the union member.
178	"""
179
180	files: tuple[str, ...]	10✔
181
182
183	@memoized	10✔
184	def _single_partition_field_set_rules(cls) -> Iterable:	10✔
185	"""Returns a rule that implements a "partitioner" for `PartitionFieldSetsRequest`, which returns
186	one partition."""
187
188	@rule(	10✔
189	canonical_name_suffix=cls.__name__,
190	_param_type_overrides={
191	"request": cls.PartitionRequest,
192	"subsystem": cls.tool_subsystem,
193	},
194	)
195	async def partitioner(	10✔
196	request: _PartitionFieldSetsRequestBase, subsystem: SkippableSubsystem
197	) -> Partitions[FieldSet, _EmptyMetadata]:
198	return Partitions() if subsystem.skip else Partitions.single_partition(request.field_sets)	×
199
200	return collect_rules(locals())	10✔
201
202
203	@memoized	10✔
204	def _single_partition_file_rules(cls) -> Iterable:	10✔
205	"""Returns a rule that implements a "partitioner" for `PartitionFieldSetsRequest`, which returns
206	one partition."""
207
208	# NB: This only works if the FieldSet has a single `SourcesField` field. We check here for
209	# a better user experience.
210	sources_field_name = _get_sources_field_name(cls.field_set_type)	9✔
211
212	@rule(	9✔
213	canonical_name_suffix=cls.__name__,
214	_param_type_overrides={
215	"request": cls.PartitionRequest,
216	"subsystem": cls.tool_subsystem,
217	},
218	)
219	async def partitioner(	9✔
220	request: _PartitionFieldSetsRequestBase, subsystem: SkippableSubsystem
221	) -> Partitions[str, _EmptyMetadata]:
222	assert sources_field_name is not None	×
223	all_sources_paths = await concurrently(	×
224	resolve_source_paths(
225	SourcesPathsRequest(getattr(field_set, sources_field_name)), **implicitly()
226	)
227	for field_set in request.field_sets
228	)
229
230	return (	×
231	Partitions()
232	if subsystem.skip
233	else Partitions.single_partition(
234	itertools.chain.from_iterable(
235	sources_paths.files for sources_paths in all_sources_paths
236	)
237	)
238	)
239
240	return collect_rules(locals())	9✔
241
242
243	@memoized	10✔
244	def _partition_per_input_field_set_rules(cls) -> Iterable:	10✔
245	"""Returns a rule that implements a "partitioner" for `PartitionFieldSetsRequest`, which returns
246	a single-element partition per input."""
247
248	@rule(	9✔
249	canonical_name_suffix=cls.__name__,
250	_param_type_overrides={
251	"request": cls.PartitionRequest,
252	"subsystem": cls.tool_subsystem,
253	},
254	)
255	async def partitioner(	9✔
256	request: _PartitionFieldSetsRequestBase, subsystem: SkippableSubsystem
257	) -> Partitions[FieldSet, _EmptyMetadata]:
258	return (	×
259	Partitions()
260	if subsystem.skip
261	else Partitions(
262	Partition((field_set,), _EmptyMetadata()) for field_set in request.field_sets
263	)
264	)
265
266	return collect_rules(locals())	9✔
267
268
269	@memoized	10✔
270	def _partition_per_input_file_rules(cls) -> Iterable:	10✔
271	"""Returns a rule that implements a "partitioner" for `PartitionFieldSetsRequest`, which returns
272	a single-element partition per input."""
273
274	# NB: This only works if the FieldSet has a single `SourcesField` field. We check here for
275	# a better user experience.
276	sources_field_name = _get_sources_field_name(cls.field_set_type)	1✔
277
278	@rule(	1✔
279	canonical_name_suffix=cls.__name__,
280	_param_type_overrides={
281	"request": cls.PartitionRequest,
282	"subsystem": cls.tool_subsystem,
283	},
284	)
285	async def partitioner(	1✔
286	request: _PartitionFieldSetsRequestBase, subsystem: SkippableSubsystem
287	) -> Partitions[str, _EmptyMetadata]:
288	assert sources_field_name is not None	×
289	all_sources_paths = await concurrently(	×
290	resolve_source_paths(
291	SourcesPathsRequest(getattr(field_set, sources_field_name)), **implicitly()
292	)
293	for field_set in request.field_sets
294	)
295
296	return (	×
297	Partitions()
298	if subsystem.skip
299	else Partitions(
300	Partition((path,), _EmptyMetadata())
301	for path in itertools.chain.from_iterable(
302	sources_paths.files for sources_paths in all_sources_paths
303	)
304	)
305	)
306
307	return collect_rules(locals())	1✔
308
309
310	def _get_sources_field_name(field_set_type: type[FieldSet]) -> str:	10✔
311	"""Get the name of the one `SourcesField` belonging to the given target type.
312
313	NOTE: The input target type's fields must contain exactly one `SourcesField`.
314	Otherwise this method will raise a `TypeError`.
315	"""
316
317	sources_field_name = None	9✔
318	for fieldname, fieldtype in field_set_type.fields.items():	9✔
319	if issubclass(fieldtype, SourcesField):	9✔
320	if sources_field_name is None:	9✔
321	sources_field_name = fieldname	9✔
322	break	9✔
323	raise TypeError(	×
324	f"Type {field_set_type} has multiple `SourcesField` fields."
325	+ " Pants can't provide a default partitioner."
326	)
327	else:
328	raise TypeError(	×
329	f"Type {field_set_type} has does not have a `SourcesField` field."
330	+ " Pants can't provide a default partitioner."
331	)
332	return sources_field_name	9✔

pantsbuild / pants / 18198316586

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