22831682209

Committed 08 Mar 2026 10:49PM UTC coverage: 92.936% (+0.007%) from 92.929%

Build # 22831682209

Build Type

Pull #23160

github

Committed by

web-flow

Commit Message

Merge 1a595ab34 into 510aad6c0

Pull Request Pull Request #23160: [pantsng] Pythonic access to the rust source partition logic.

Run Details

150 of 157 new or added lines in 2 files covered. (95.54%)

9 existing lines in 1 file now uncovered.

91109 of 98034 relevant lines covered (92.94%)

4.38 hits per line

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

89.06

/src/python/pants/ng/source_partition.py

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

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path

from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior
from pants.engine.collection import Collection
from pants.engine.fs import GlobExpansionConjunction, PathGlobs
from pants.engine.internals.native_engine import (
    Digest,
    PyNgOptions,
    PyNgOptionsReader,
    PyNgSourcePartition,
)
from pants.engine.internals.session import SessionValues
from pants.engine.intrinsics import path_globs_to_digest, path_globs_to_paths
from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, implicitly, rule
from pants.source.source_root import SourceRoot, SourceRootsRequest, get_source_roots
from pants.util.memo import memoized_property


@dataclass(frozen=True)
class SourcePaths:
    """A set of sources, under a single source root."""

    paths: tuple[Path, ...]
    source_root: SourceRoot

    def path_strs(self) -> tuple[str, ...]:
        return tuple(str(path) for path in self.paths)

    def commondir(self) -> str:
        ret = os.path.commonpath(self.paths)
        if os.path.isfile(ret):
            ret = os.path.dirname(ret)
        return ret

    def filter_by_suffixes(self, suffixes: tuple[str, ...]) -> SourcePaths:
        suffixes_set = set(suffixes)
        return SourcePaths(
            tuple(path for path in self.paths if path.suffix in suffixes_set),
            self.source_root,
        )


@dataclass(frozen=True)
class SourceDigest:
    digest: Digest


@rule
async def source_paths_to_digest(source_paths: SourcePaths) -> SourceDigest:
    source_digest = await path_globs_to_digest(
        PathGlobs(
            source_paths.path_strs(),
            glob_match_error_behavior=GlobMatchErrorBehavior.error,
            conjunction=GlobExpansionConjunction.all_match,
            description_of_origin="Input source paths",
        )
    )
    return SourceDigest(source_digest)


@dataclass(frozen=True)
class SourcePartition:
    """Access to source files and the config that goes with them."""

    _native_partition: PyNgSourcePartition
    _source_root: SourceRoot

    @memoized_property
    def source_paths(self) -> SourcePaths:
        return SourcePaths(
            tuple(Path(p) for p in self._native_partition.paths()), self._source_root
        )

    @memoized_property
    def options_reader(self) -> PyNgOptionsReader:
        return self._native_partition.options_reader()


class SourcePartitions(Collection[SourcePartition]):
    pass


# Uncacheable because we must get the most recent session value on each run.
@_uncacheable_rule
async def get_ng_options(session_values: SessionValues) -> PyNgOptions:
    return session_values[PyNgOptions]


# Uncacheable because we must recompute on each run.
@_uncacheable_rule
async def partition_sources(path_globs: PathGlobs) -> SourcePartitions:
    options = await get_ng_options(**implicitly())
    paths = await path_globs_to_paths(path_globs)
    # First partition by source root.
    source_roots = await get_source_roots(SourceRootsRequest.for_files(paths.files))
    root_to_paths = source_roots.root_to_paths()
    partitions: list[SourcePartition] = []
    for source_root, paths_in_partition in root_to_paths.items():
        # Then subpartition each of those by config.
        partitions.extend(
            SourcePartition(native_part, source_root)
            for native_part in options.partition_sources(
                tuple(str(path) for path in paths_in_partition)
            )
        )
    return SourcePartitions(tuple(partitions))


@rule
async def get_source_paths(partition: SourcePartition) -> SourcePaths:
    return partition.source_paths


def rules() -> tuple[Rule, ...]:
    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	1✔
5
6	import os	1✔
7	from dataclasses import dataclass	1✔
8	from pathlib import Path	1✔
9
10	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	1✔
11	from pants.engine.collection import Collection	1✔
12	from pants.engine.fs import GlobExpansionConjunction, PathGlobs	1✔
13	from pants.engine.internals.native_engine import (	1✔
14	Digest,
15	PyNgOptions,
16	PyNgOptionsReader,
17	PyNgSourcePartition,
18	)
19	from pants.engine.internals.session import SessionValues	1✔
20	from pants.engine.intrinsics import path_globs_to_digest, path_globs_to_paths	1✔
21	from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, implicitly, rule	1✔
22	from pants.source.source_root import SourceRoot, SourceRootsRequest, get_source_roots	1✔
23	from pants.util.memo import memoized_property	1✔
24
25
26	@dataclass(frozen=True)	1✔
27	class SourcePaths:	1✔
28	"""A set of sources, under a single source root."""
29
30	paths: tuple[Path, ...]	1✔
31	source_root: SourceRoot	1✔
32
33	def path_strs(self) -> tuple[str, ...]:	1✔
NEW 34	return tuple(str(path) for path in self.paths)	×
35
36	def commondir(self) -> str:	1✔
37	ret = os.path.commonpath(self.paths)	1✔
38	if os.path.isfile(ret):	1✔
39	ret = os.path.dirname(ret)	1✔
40	return ret	1✔
41
42	def filter_by_suffixes(self, suffixes: tuple[str, ...]) -> SourcePaths:	1✔
43	suffixes_set = set(suffixes)	1✔
44	return SourcePaths(	1✔
45	tuple(path for path in self.paths if path.suffix in suffixes_set),
46	self.source_root,
47	)
48
49
50	@dataclass(frozen=True)	1✔
51	class SourceDigest:	1✔
52	digest: Digest	1✔
53
54
55	@rule	1✔
56	async def source_paths_to_digest(source_paths: SourcePaths) -> SourceDigest:	1✔
NEW 57	source_digest = await path_globs_to_digest(	×
58	PathGlobs(
59	source_paths.path_strs(),
60	glob_match_error_behavior=GlobMatchErrorBehavior.error,
61	conjunction=GlobExpansionConjunction.all_match,
62	description_of_origin="Input source paths",
63	)
64	)
NEW 65	return SourceDigest(source_digest)	×
66
67
68	@dataclass(frozen=True)	1✔
69	class SourcePartition:	1✔
70	"""Access to source files and the config that goes with them."""
71
72	_native_partition: PyNgSourcePartition	1✔
73	_source_root: SourceRoot	1✔
74
75	@memoized_property	1✔
76	def source_paths(self) -> SourcePaths:	1✔
77	return SourcePaths(	1✔
78	tuple(Path(p) for p in self._native_partition.paths()), self._source_root
79	)
80
81	@memoized_property	1✔
82	def options_reader(self) -> PyNgOptionsReader:	1✔
NEW 83	return self._native_partition.options_reader()	×
84
85
86	class SourcePartitions(Collection[SourcePartition]):	1✔
87	pass	1✔
88
89
90	# Uncacheable because we must get the most recent session value on each run.
91	@_uncacheable_rule	1✔
92	async def get_ng_options(session_values: SessionValues) -> PyNgOptions:	1✔
NEW 93	return session_values[PyNgOptions]	×
94
95
96	# Uncacheable because we must recompute on each run.
97	@_uncacheable_rule	1✔
98	async def partition_sources(path_globs: PathGlobs) -> SourcePartitions:	1✔
99	options = await get_ng_options(**implicitly())	1✔
100	paths = await path_globs_to_paths(path_globs)	1✔
101	# First partition by source root.
102	source_roots = await get_source_roots(SourceRootsRequest.for_files(paths.files))	1✔
103	root_to_paths = source_roots.root_to_paths()	1✔
104	partitions: list[SourcePartition] = []	1✔
105	for source_root, paths_in_partition in root_to_paths.items():	1✔
106	# Then subpartition each of those by config.
107	partitions.extend(	1✔
108	SourcePartition(native_part, source_root)
109	for native_part in options.partition_sources(
110	tuple(str(path) for path in paths_in_partition)
111	)
112	)
113	return SourcePartitions(tuple(partitions))	1✔
114
115
116	@rule	1✔
117	async def get_source_paths(partition: SourcePartition) -> SourcePaths:	1✔
NEW 118	return partition.source_paths	×
119
120
121	def rules() -> tuple[Rule, ...]:	1✔
NEW 122	return (*collect_rules(),)	×

pantsbuild / pants / 22831682209

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