19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

72.94

/src/python/pants/engine/internals/synthetic_targets.py

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

"""Synthetic targets is a concept of injecting targets into the build graph that doesn't have a home
in any BUILD file.

This is achieved by declaring a union member for the `SyntheticTargetsRequest` union along
with a rule taking that member and returning a collection of `SyntheticAddressMap`s.

Example demonstrating how to register synthetic targets:

    from dataclasses import dataclass
    from pants.engine.internals.synthetic_targets import (
        SyntheticAddressMaps,
        SyntheticTargetsRequest,
        SyntheticTargetsSpecPaths,
    )
    from pants.engine.internals.target_adaptor import TargetAdaptor
    from pants.engine.rules import collect_rules, rule


    @dataclass(frozen=True)
    class SyntheticExampleTargetsRequest(SyntheticTargetsRequest):
        path: str = SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS


    class SyntheticExampleTargetsPerDirectorySpecPathsRequest:
        pass


    @dataclass(frozen=True)
    class SyntheticExampleTargetsPerDirectoryRequest(SyntheticTargetsRequest):
        path: str = SyntheticTargetsRequest.REQUEST_TARGETS_PER_DIRECTORY

        # Optional: (without it, only paths with BUILD files will be consistently considered)
        spec_paths_request = SyntheticExampleTargetsPerDirectorySpecPathsRequest


    @rule
    async def example_synthetic_targets_per_directory_spec_paths(
        request: SyntheticExampleTargetsPerDirectorySpecPathsRequest,
    ) -> SyntheticTargetsSpecPaths:
        # Return all paths we have targets for.
        # This may involve using GlobPaths etc to discover files in the project source tree.
        known_paths = ["src/a/dir1", "src/a/dir2", ...]
        return SyntheticTargetsSpecPaths.from_paths(known_paths)


    @rule
    async def example_synthetic_targets(request: SyntheticExampleTargetsRequest) -> SyntheticAddressMaps:
        # Return targets for `request.path`.
        return SyntheticAddressMaps.for_targets_request(
            request,
            [
                (
                  "BUILD.synthetic-example",
                  (
                    TargetAdaptor("<type-alias>", "<name>", description_of_origin="...", **kwargs),
                    ...
                  ),
                ),
                ...
            ]
        )


    def rules():
        return (
            *collect_rules(),
            SyntheticExampleTargetsRequest.rules(),
            SyntheticExampleTargetsPerDirectoryRequest.rules(),
            ...
        )
"""

from __future__ import annotations

import itertools
import os.path
from collections.abc import Iterable, Iterator, Sequence
from dataclasses import dataclass
from typing import ClassVar

from pants.base.specs import GlobSpecsProtocol
from pants.engine.collection import Collection
from pants.engine.internals.mapper import AddressMap
from pants.engine.internals.selectors import concurrently
from pants.engine.internals.target_adaptor import TargetAdaptor
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.unions import UnionMembership, UnionRule, union
from pants.util.frozendict import FrozenDict


@dataclass(frozen=True)
class SyntheticTargetsSpecPathsRequest:
    specs: tuple[GlobSpecsProtocol, ...]


class SyntheticTargetsSpecPaths(Collection[str]):
    @staticmethod
    def from_paths(paths: Iterable[str]) -> SyntheticTargetsSpecPaths:
        return SyntheticTargetsSpecPaths(sorted(set(paths)))

    @staticmethod
    def from_request(
        request: SyntheticTargetsSpecPathsRequest, paths: Iterable[str]
    ) -> SyntheticTargetsSpecPaths:
        return SyntheticTargetsSpecPaths.from_paths(
            filter(
                lambda path: any(spec.matches_target_residence_dir(path) for spec in request.specs),
                paths,
            )
        )


@union
class _SpecPathsRequest:
    """Protected union type."""


@rule(polymorphic=True)
async def _get_spec_paths(req: _SpecPathsRequest) -> SyntheticTargetsSpecPaths:
    raise NotImplementedError()


@union
@dataclass(frozen=True)
class SyntheticTargetsRequest:
    """Union members of the `SyntheticTargetsRequest` should implement a rule returning an instance
    of a `SyntheticAddressMaps`."""

    REQUEST_TARGETS_PER_DIRECTORY: ClassVar[str] = "*"
    SINGLE_REQUEST_FOR_ALL_TARGETS: ClassVar[str] = ""

    path: str = REQUEST_TARGETS_PER_DIRECTORY
    """
    The default field value is used to filter which paths to request targets for, and should be
    declared as appropriate by union members subclassing `SyntheticTargetsRequest`. The
    SINGLE_REQUEST_FOR_ALL_TARGETS will make a single request for all targets at once, while
    REQUEST_TARGETS_PER_DIRECTORY will request all targets for a particular path at a time. Any
    other value will be used as filter to only request targets for that specific directory.

    The rule processing this request should inspect `request.path` to only include targets for that
    directory, unless `request.path` is `SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS` in
    which case _all_ synthetic targets should be returned.
    """

    spec_paths_request: ClassVar[type | None] = None
    """Request class for providing paths in addition to those where BUILD files are found.

    Implement a rule that takes `spec_paths_request` and returns an `SyntheticTargetsSpecPaths`.
    """

    @classmethod
    def rules(cls) -> Iterator[UnionRule]:
        yield UnionRule(SyntheticTargetsRequest, cls)
        if cls.spec_paths_request is not None:
            yield UnionRule(_SpecPathsRequest, cls.spec_paths_request)


class SyntheticAddressMap(AddressMap):
    pass


@dataclass(frozen=True)
class SyntheticAddressMapsRequest:
    """Request all registered synthetic targets for a given path."""

    path: str


class SyntheticAddressMaps(Collection[SyntheticAddressMap]):
    """A collection of `SyntheticAddressMap` for all synthetic target adaptors."""

    @classmethod
    def for_targets_request(
        cls,
        request: SyntheticTargetsRequest,
        synthetic_target_adaptors: Iterable[tuple[str, Iterable[TargetAdaptor]]],
    ) -> SyntheticAddressMaps:
        return cls(
            SyntheticAddressMap.create(os.path.join(request.path, filename), target_adaptors)
            for filename, target_adaptors in synthetic_target_adaptors
        )


@rule(polymorphic=True)
async def _get_synthetic_address_maps(req: SyntheticTargetsRequest) -> SyntheticAddressMaps:
    raise NotImplementedError()


@dataclass(frozen=True)
class AllSyntheticAddressMaps:
    """All pre-loaded SyntheticAddressMaps per directory."""

    address_maps: FrozenDict[str, SyntheticAddressMaps]
    path_request_types: FrozenDict[str, Sequence[type[SyntheticTargetsRequest]]]
    spec_paths: Sequence[str]

    @classmethod
    def create(
        cls,
        address_maps: Iterable[SyntheticAddressMap],
        requests: Iterable[SyntheticTargetsRequest],
        spec_paths: Iterable[str],
    ) -> AllSyntheticAddressMaps:
        def address_map_key(address_map: SyntheticAddressMap) -> str:
            return os.path.dirname(address_map.path)

        def requests_key(request: SyntheticAddressMapsRequest) -> str:
            return request.path

        return AllSyntheticAddressMaps(
            address_maps=FrozenDict(
                {
                    path: SyntheticAddressMaps(address_maps_group)
                    for path, address_maps_group in itertools.groupby(
                        sorted(address_maps, key=address_map_key), key=address_map_key
                    )
                }
            ),
            path_request_types=FrozenDict(
                {
                    path: tuple(type(request) for request in requests_group)  # type: ignore[misc]
                    for path, requests_group in itertools.groupby(
                        sorted(requests, key=requests_key),  # type: ignore[arg-type]
                        key=requests_key,  # type: ignore[arg-type]
                    )
                    if path != SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS
                }
            ),
            spec_paths=tuple(sorted(spec_paths)),
        )

    def targets_request_types(self, path: str) -> Iterable[type[SyntheticTargetsRequest]]:
        yield from self.path_request_types.get(
            SyntheticTargetsRequest.REQUEST_TARGETS_PER_DIRECTORY, ()
        )
        yield from self.path_request_types.get(path, ())


@rule
async def get_synthetic_address_maps(
    request: SyntheticAddressMapsRequest,
    all_synthetic: AllSyntheticAddressMaps,
) -> SyntheticAddressMaps:
    per_directory_address_maps = await concurrently(
        _get_synthetic_address_maps(
            **implicitly({request_type(request.path): SyntheticTargetsRequest})
        )
        for request_type in all_synthetic.targets_request_types(request.path)
    )

    return SyntheticAddressMaps(
        itertools.chain(
            all_synthetic.address_maps.get(request.path, ()), *per_directory_address_maps
        )
    )


@rule
async def all_synthetic_targets(union_membership: UnionMembership) -> AllSyntheticAddressMaps:
    requests = [request_type() for request_type in union_membership.get(SyntheticTargetsRequest)]
    all_synthetic = await concurrently(
        _get_synthetic_address_maps(**implicitly({request: SyntheticTargetsRequest}))
        for request in requests
        if request.path == SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS
    )
    all_spec_paths = await concurrently(
        _get_spec_paths(
            **implicitly({spec_paths_request(): _SpecPathsRequest}),
        )
        for spec_paths_request in union_membership.get(_SpecPathsRequest)
    )
    return AllSyntheticAddressMaps.create(
        address_maps=itertools.chain.from_iterable(all_synthetic),
        requests=requests,
        spec_paths=set(itertools.chain.from_iterable(all_spec_paths)),
    )


@rule
async def get_synthetic_targets_spec_paths(
    request: SyntheticTargetsSpecPathsRequest, all_synthetic: AllSyntheticAddressMaps
) -> SyntheticTargetsSpecPaths:
    """Return all known paths for synthetic targets."""
    return SyntheticTargetsSpecPaths.from_request(
        request, itertools.chain(all_synthetic.address_maps, all_synthetic.spec_paths)
    )


def rules():
    return collect_rules()

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	"""Synthetic targets is a concept of injecting targets into the build graph that doesn't have a home
5	in any BUILD file.
6
7	This is achieved by declaring a union member for the `SyntheticTargetsRequest` union along
8	with a rule taking that member and returning a collection of `SyntheticAddressMap`s.
9
10	Example demonstrating how to register synthetic targets:
11
12	from dataclasses import dataclass
13	from pants.engine.internals.synthetic_targets import (
14	SyntheticAddressMaps,
15	SyntheticTargetsRequest,
16	SyntheticTargetsSpecPaths,
17	)
18	from pants.engine.internals.target_adaptor import TargetAdaptor
19	from pants.engine.rules import collect_rules, rule
20
21
22	@dataclass(frozen=True)
23	class SyntheticExampleTargetsRequest(SyntheticTargetsRequest):
24	path: str = SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS
25
26
27	class SyntheticExampleTargetsPerDirectorySpecPathsRequest:
28	pass
29
30
31	@dataclass(frozen=True)
32	class SyntheticExampleTargetsPerDirectoryRequest(SyntheticTargetsRequest):
33	path: str = SyntheticTargetsRequest.REQUEST_TARGETS_PER_DIRECTORY
34
35	# Optional: (without it, only paths with BUILD files will be consistently considered)
36	spec_paths_request = SyntheticExampleTargetsPerDirectorySpecPathsRequest
37
38
39	@rule
40	async def example_synthetic_targets_per_directory_spec_paths(
41	request: SyntheticExampleTargetsPerDirectorySpecPathsRequest,
42	) -> SyntheticTargetsSpecPaths:
43	# Return all paths we have targets for.
44	# This may involve using GlobPaths etc to discover files in the project source tree.
45	known_paths = ["src/a/dir1", "src/a/dir2", ...]
46	return SyntheticTargetsSpecPaths.from_paths(known_paths)
47
48
49	@rule
50	async def example_synthetic_targets(request: SyntheticExampleTargetsRequest) -> SyntheticAddressMaps:
51	# Return targets for `request.path`.
52	return SyntheticAddressMaps.for_targets_request(
53	request,
54	[
55	(
56	"BUILD.synthetic-example",
57	(
58	TargetAdaptor("<type-alias>", "<name>", description_of_origin="...", **kwargs),
59	...
60	),
61	),
62	...
63	]
64	)
65
66
67	def rules():
68	return (
69	*collect_rules(),
70	SyntheticExampleTargetsRequest.rules(),
71	SyntheticExampleTargetsPerDirectoryRequest.rules(),
72	...
73	)
74	"""
75
76	from __future__ import annotations	1✔
77
78	import itertools	1✔
79	import os.path	1✔
80	from collections.abc import Iterable, Iterator, Sequence	1✔
81	from dataclasses import dataclass	1✔
82	from typing import ClassVar	1✔
83
84	from pants.base.specs import GlobSpecsProtocol	1✔
85	from pants.engine.collection import Collection	1✔
86	from pants.engine.internals.mapper import AddressMap	1✔
87	from pants.engine.internals.selectors import concurrently	1✔
88	from pants.engine.internals.target_adaptor import TargetAdaptor	1✔
89	from pants.engine.rules import collect_rules, implicitly, rule	1✔
90	from pants.engine.unions import UnionMembership, UnionRule, union	1✔
91	from pants.util.frozendict import FrozenDict	1✔
92
93
94	@dataclass(frozen=True)	1✔
95	class SyntheticTargetsSpecPathsRequest:	1✔
96	specs: tuple[GlobSpecsProtocol, ...]	1✔
97
98
99	class SyntheticTargetsSpecPaths(Collection[str]):	1✔
100	@staticmethod	1✔
101	def from_paths(paths: Iterable[str]) -> SyntheticTargetsSpecPaths:	1✔
102	return SyntheticTargetsSpecPaths(sorted(set(paths)))	×
103
104	@staticmethod	1✔
105	def from_request(	1✔
106	request: SyntheticTargetsSpecPathsRequest, paths: Iterable[str]
107	) -> SyntheticTargetsSpecPaths:
108	return SyntheticTargetsSpecPaths.from_paths(	×
109	filter(
110	lambda path: any(spec.matches_target_residence_dir(path) for spec in request.specs),
111	paths,
112	)
113	)
114
115
116	@union	1✔
117	class _SpecPathsRequest:	1✔
118	"""Protected union type."""
119
120
121	@rule(polymorphic=True)	1✔
122	async def _get_spec_paths(req: _SpecPathsRequest) -> SyntheticTargetsSpecPaths:	1✔
123	raise NotImplementedError()	×
124
125
126	@union	1✔
127	@dataclass(frozen=True)	1✔
128	class SyntheticTargetsRequest:	1✔
129	"""Union members of the `SyntheticTargetsRequest` should implement a rule returning an instance
130	of a `SyntheticAddressMaps`."""
131
132	REQUEST_TARGETS_PER_DIRECTORY: ClassVar[str] = "*"	1✔
133	SINGLE_REQUEST_FOR_ALL_TARGETS: ClassVar[str] = ""	1✔
134
135	path: str = REQUEST_TARGETS_PER_DIRECTORY	1✔
136	"""	1✔
137	The default field value is used to filter which paths to request targets for, and should be
138	declared as appropriate by union members subclassing `SyntheticTargetsRequest`. The
139	SINGLE_REQUEST_FOR_ALL_TARGETS will make a single request for all targets at once, while
140	REQUEST_TARGETS_PER_DIRECTORY will request all targets for a particular path at a time. Any
141	other value will be used as filter to only request targets for that specific directory.
142
143	The rule processing this request should inspect `request.path` to only include targets for that
144	directory, unless `request.path` is `SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS` in
145	which case _all_ synthetic targets should be returned.
146	"""
147
148	spec_paths_request: ClassVar[type \| None] = None	1✔
149	"""Request class for providing paths in addition to those where BUILD files are found.	1✔
150
151	Implement a rule that takes `spec_paths_request` and returns an `SyntheticTargetsSpecPaths`.
152	"""
153
154	@classmethod	1✔
155	def rules(cls) -> Iterator[UnionRule]:	1✔
UNCOV 156	yield UnionRule(SyntheticTargetsRequest, cls)	×
UNCOV 157	if cls.spec_paths_request is not None:	×
UNCOV 158	yield UnionRule(_SpecPathsRequest, cls.spec_paths_request)	×
159
160
161	class SyntheticAddressMap(AddressMap):	1✔
162	pass	1✔
163
164
165	@dataclass(frozen=True)	1✔
166	class SyntheticAddressMapsRequest:	1✔
167	"""Request all registered synthetic targets for a given path."""
168
169	path: str	1✔
170
171
172	class SyntheticAddressMaps(Collection[SyntheticAddressMap]):	1✔
173	"""A collection of `SyntheticAddressMap` for all synthetic target adaptors."""
174
175	@classmethod	1✔
176	def for_targets_request(	1✔
177	cls,
178	request: SyntheticTargetsRequest,
179	synthetic_target_adaptors: Iterable[tuple[str, Iterable[TargetAdaptor]]],
180	) -> SyntheticAddressMaps:
181	return cls(	×
182	SyntheticAddressMap.create(os.path.join(request.path, filename), target_adaptors)
183	for filename, target_adaptors in synthetic_target_adaptors
184	)
185
186
187	@rule(polymorphic=True)	1✔
188	async def _get_synthetic_address_maps(req: SyntheticTargetsRequest) -> SyntheticAddressMaps:	1✔
189	raise NotImplementedError()	×
190
191
192	@dataclass(frozen=True)	1✔
193	class AllSyntheticAddressMaps:	1✔
194	"""All pre-loaded SyntheticAddressMaps per directory."""
195
196	address_maps: FrozenDict[str, SyntheticAddressMaps]	1✔
197	path_request_types: FrozenDict[str, Sequence[type[SyntheticTargetsRequest]]]	1✔
198	spec_paths: Sequence[str]	1✔
199
200	@classmethod	1✔
201	def create(	1✔
202	cls,
203	address_maps: Iterable[SyntheticAddressMap],
204	requests: Iterable[SyntheticTargetsRequest],
205	spec_paths: Iterable[str],
206	) -> AllSyntheticAddressMaps:
207	def address_map_key(address_map: SyntheticAddressMap) -> str:	×
208	return os.path.dirname(address_map.path)	×
209
210	def requests_key(request: SyntheticAddressMapsRequest) -> str:	×
211	return request.path	×
212
213	return AllSyntheticAddressMaps(	×
214	address_maps=FrozenDict(
215	{
216	path: SyntheticAddressMaps(address_maps_group)
217	for path, address_maps_group in itertools.groupby(
218	sorted(address_maps, key=address_map_key), key=address_map_key
219	)
220	}
221	),
222	path_request_types=FrozenDict(
223	{
224	path: tuple(type(request) for request in requests_group) # type: ignore[misc]
225	for path, requests_group in itertools.groupby(
226	sorted(requests, key=requests_key), # type: ignore[arg-type]
227	key=requests_key, # type: ignore[arg-type]
228	)
229	if path != SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS
230	}
231	),
232	spec_paths=tuple(sorted(spec_paths)),
233	)
234
235	def targets_request_types(self, path: str) -> Iterable[type[SyntheticTargetsRequest]]:	1✔
236	yield from self.path_request_types.get(	×
237	SyntheticTargetsRequest.REQUEST_TARGETS_PER_DIRECTORY, ()
238	)
239	yield from self.path_request_types.get(path, ())	×
240
241
242	@rule	1✔
243	async def get_synthetic_address_maps(	1✔
244	request: SyntheticAddressMapsRequest,
245	all_synthetic: AllSyntheticAddressMaps,
246	) -> SyntheticAddressMaps:
247	per_directory_address_maps = await concurrently(	×
248	_get_synthetic_address_maps(
249	**implicitly({request_type(request.path): SyntheticTargetsRequest})
250	)
251	for request_type in all_synthetic.targets_request_types(request.path)
252	)
253
254	return SyntheticAddressMaps(	×
255	itertools.chain(
256	all_synthetic.address_maps.get(request.path, ()), *per_directory_address_maps
257	)
258	)
259
260
261	@rule	1✔
262	async def all_synthetic_targets(union_membership: UnionMembership) -> AllSyntheticAddressMaps:	1✔
263	requests = [request_type() for request_type in union_membership.get(SyntheticTargetsRequest)]	×
264	all_synthetic = await concurrently(	×
265	_get_synthetic_address_maps(**implicitly({request: SyntheticTargetsRequest}))
266	for request in requests
267	if request.path == SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS
268	)
269	all_spec_paths = await concurrently(	×
270	_get_spec_paths(
271	**implicitly({spec_paths_request(): _SpecPathsRequest}),
272	)
273	for spec_paths_request in union_membership.get(_SpecPathsRequest)
274	)
275	return AllSyntheticAddressMaps.create(	×
276	address_maps=itertools.chain.from_iterable(all_synthetic),
277	requests=requests,
278	spec_paths=set(itertools.chain.from_iterable(all_spec_paths)),
279	)
280
281
282	@rule	1✔
283	async def get_synthetic_targets_spec_paths(	1✔
284	request: SyntheticTargetsSpecPathsRequest, all_synthetic: AllSyntheticAddressMaps
285	) -> SyntheticTargetsSpecPaths:
286	"""Return all known paths for synthetic targets."""
287	return SyntheticTargetsSpecPaths.from_request(	×
288	request, itertools.chain(all_synthetic.address_maps, all_synthetic.spec_paths)
289	)
290
291
292	def rules():	1✔
UNCOV 293	return collect_rules()	×

pantsbuild / pants / 19015773527

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