25443604553

Committed 06 May 2026 03:05PM UTC coverage: 92.879% (-0.04%) from 92.915%

Build # 25443604553

Build Type

push

github

Committed by

web-flow

Commit Message

[pants_ng] Scaffolding for a pants_ng mode. (#23319)

In this mode the command line is parsed as an
NG invocation, and dispatched appropriately.

Of course at the moment there are no
implementations to dispatch to. That will follow.

This does expose a new option, `pants_ng` to users. 
There is a big warning not to set it, but we're not trying
to hide that we're working on a new thing, so I am
comfortable with this.

Coverage Stats

25 of 76 new or added lines in 9 files covered. (32.89%)

1294 existing lines in 76 files now uncovered.

92234 of 99306 relevant lines covered (92.88%)

4.05 hits per line

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

99.05

/src/python/pants/engine/intrinsics.py

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

from __future__ import annotations

import dataclasses
import logging

from pants.engine.environment import EnvironmentName
from pants.engine.fs import (
    AddPrefix,
    CreateDigest,
    Digest,
    DigestContents,
    DigestEntries,
    DigestSubset,
    MergeDigests,
    NativeDownloadFile,
    PathGlobs,
    PathMetadataRequest,
    PathMetadataResult,
    Paths,
    RemovePrefix,
    Snapshot,
)
from pants.engine.internals import native_engine
from pants.engine.internals.docker import DockerResolveImageRequest, DockerResolveImageResult
from pants.engine.internals.native_dep_inference import (
    NativeDockerfileInfos,
    NativeJavascriptFilesDependencies,
    NativePythonFilesDependencies,
)
from pants.engine.internals.native_engine import NativeDependenciesRequest, task_side_effected
from pants.engine.internals.session import RunId, SessionValues
from pants.engine.process import (
    FallibleProcessResult,
    InteractiveProcess,
    InteractiveProcessResult,
    Process,
    ProcessExecutionEnvironment,
    ProcessResultWithRetries,
    ProcessWithRetries,
)
from pants.engine.rules import _uncacheable_rule, collect_rules, implicitly, rule
from pants.util.docutil import git_url
from pants.util.frozendict import FrozenDict


@rule
async def create_digest(
    create_digest: CreateDigest,
) -> Digest:
    return await native_engine.create_digest(create_digest)


@rule
async def path_globs_to_digest(
    path_globs: PathGlobs,
) -> Digest:
    return await native_engine.path_globs_to_digest(path_globs)


@rule
async def path_globs_to_paths(
    path_globs: PathGlobs,
) -> Paths:
    return await native_engine.path_globs_to_paths(path_globs)


@rule
async def download_file(
    native_download_file: NativeDownloadFile,
) -> Digest:
    return await native_engine.download_file(native_download_file)


@rule
async def digest_to_snapshot(digest: Digest) -> Snapshot:
    return await native_engine.digest_to_snapshot(digest)


@rule
async def get_digest_contents(digest: Digest) -> DigestContents:
    return await native_engine.get_digest_contents(digest)


@rule
async def get_digest_entries(digest: Digest) -> DigestEntries:
    return await native_engine.get_digest_entries(digest)


@rule
async def merge_digests(merge_digests: MergeDigests) -> Digest:
    return await native_engine.merge_digests(merge_digests)


@rule
async def remove_prefix(remove_prefix: RemovePrefix) -> Digest:
    return await native_engine.remove_prefix(remove_prefix)


@rule
async def add_prefix(add_prefix: AddPrefix) -> Digest:
    return await native_engine.add_prefix(add_prefix)


@rule
async def execute_process(
    process: Process, process_execution_environment: ProcessExecutionEnvironment
) -> FallibleProcessResult:
    return await native_engine.execute_process(process, process_execution_environment)


@rule
async def execute_process_with_retry(req: ProcessWithRetries) -> ProcessResultWithRetries:
    results: list[FallibleProcessResult] = []
    for attempt in range(0, req.attempts):
        proc = dataclasses.replace(req.proc, attempt=attempt)
        result = await execute_process(proc, **implicitly())
        results.append(result)
        if result.exit_code == 0:
            break
    return ProcessResultWithRetries(tuple(results))


@rule
async def digest_subset_to_digest(digest_subset: DigestSubset) -> Digest:
    return await native_engine.digest_subset_to_digest(digest_subset)


@rule
async def session_values() -> SessionValues:
    return await native_engine.session_values()


@rule
async def run_id() -> RunId:
    return await native_engine.run_id()


__SQUELCH_WARNING = "__squelch_warning"


# NB: Call one of the helpers below, instead of calling this rule directly,
#  to ensure correct application of restartable logic.
@_uncacheable_rule
async def _interactive_process(
    process: InteractiveProcess, process_execution_environment: ProcessExecutionEnvironment
) -> InteractiveProcessResult:
    # This is a crafty way for a caller to signal into this function without a dedicated arg
    # (which would confound the solver).  Note that we go via __dict__ instead of using
    # setattr/delattr, because those error for frozen dataclasses.
    if __SQUELCH_WARNING in process.__dict__:
        del process.__dict__[__SQUELCH_WARNING]
    else:
        logging.warning(
            "A plugin is calling `await _interactive_process(...)` directly. This will cause "
            "restarting logic not to be applied. Use `await run_interactive_process(process)` "
            "or `await run_interactive_process_in_environment(process, environment_name)` instead. "
            f"See {git_url('src/python/pants/engine/intrinsics.py')} for more details."
        )
    return await native_engine.interactive_process(process, process_execution_environment)


async def run_interactive_process(process: InteractiveProcess) -> InteractiveProcessResult:
    # NB: We must call task_side_effected() in this helper, rather than in a nested @rule call,
    #  so that the Task for the @rule that calls this helper is the one marked as non-restartable.
    if not process.restartable:
        task_side_effected()

    process.__dict__[__SQUELCH_WARNING] = True
    ret: InteractiveProcessResult = await _interactive_process(process, **implicitly())
    return ret


async def run_interactive_process_in_environment(
    process: InteractiveProcess, environment_name: EnvironmentName
) -> InteractiveProcessResult:
    # NB: We must call task_side_effected() in this helper, rather than in a nested @rule call,
    #  so that the Task for the @rule that calls this helper is the one marked as non-restartable.
    if not process.restartable:
        task_side_effected()

    process.__dict__[__SQUELCH_WARNING] = True
    ret: InteractiveProcessResult = await _interactive_process(
        process, **implicitly({environment_name: EnvironmentName})
    )
    return ret


@rule
async def docker_resolve_image(request: DockerResolveImageRequest) -> DockerResolveImageResult:
    return await native_engine.docker_resolve_image(request)


@rule
async def parse_dockerfile_info(
    deps_request: NativeDependenciesRequest,
) -> NativeDockerfileInfos:
    path_infos_pairs = await native_engine.parse_dockerfile_info(deps_request)
    return NativeDockerfileInfos(FrozenDict(path_infos_pairs))


@rule
async def parse_python_deps(
    deps_request: NativeDependenciesRequest,
) -> NativePythonFilesDependencies:
    path_deps_pairs = await native_engine.parse_python_deps(deps_request)
    return NativePythonFilesDependencies(FrozenDict(path_deps_pairs))


@rule
async def parse_javascript_deps(
    deps_request: NativeDependenciesRequest,
) -> NativeJavascriptFilesDependencies:
    path_deps_pairs = await native_engine.parse_javascript_deps(deps_request)
    return NativeJavascriptFilesDependencies(FrozenDict(path_deps_pairs))


@rule
async def path_metadata_request(request: PathMetadataRequest) -> PathMetadataResult:
    return await native_engine.path_metadata_request(request)


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

1	# Copyright 2024 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	12✔
5
6	import dataclasses	12✔
7	import logging	12✔
8
9	from pants.engine.environment import EnvironmentName	12✔
10	from pants.engine.fs import (	12✔
11	AddPrefix,
12	CreateDigest,
13	Digest,
14	DigestContents,
15	DigestEntries,
16	DigestSubset,
17	MergeDigests,
18	NativeDownloadFile,
19	PathGlobs,
20	PathMetadataRequest,
21	PathMetadataResult,
22	Paths,
23	RemovePrefix,
24	Snapshot,
25	)
26	from pants.engine.internals import native_engine	12✔
27	from pants.engine.internals.docker import DockerResolveImageRequest, DockerResolveImageResult	12✔
28	from pants.engine.internals.native_dep_inference import (	12✔
29	NativeDockerfileInfos,
30	NativeJavascriptFilesDependencies,
31	NativePythonFilesDependencies,
32	)
33	from pants.engine.internals.native_engine import NativeDependenciesRequest, task_side_effected	12✔
34	from pants.engine.internals.session import RunId, SessionValues	12✔
35	from pants.engine.process import (	12✔
36	FallibleProcessResult,
37	InteractiveProcess,
38	InteractiveProcessResult,
39	Process,
40	ProcessExecutionEnvironment,
41	ProcessResultWithRetries,
42	ProcessWithRetries,
43	)
44	from pants.engine.rules import _uncacheable_rule, collect_rules, implicitly, rule	12✔
45	from pants.util.docutil import git_url	12✔
46	from pants.util.frozendict import FrozenDict	12✔
47
48
49	@rule	12✔
50	async def create_digest(	12✔
51	create_digest: CreateDigest,
52	) -> Digest:
53	return await native_engine.create_digest(create_digest)	12✔
54
55
56	@rule	12✔
57	async def path_globs_to_digest(	12✔
58	path_globs: PathGlobs,
59	) -> Digest:
60	return await native_engine.path_globs_to_digest(path_globs)	12✔
61
62
63	@rule	12✔
64	async def path_globs_to_paths(	12✔
65	path_globs: PathGlobs,
66	) -> Paths:
67	return await native_engine.path_globs_to_paths(path_globs)	12✔
68
69
70	@rule	12✔
71	async def download_file(	12✔
72	native_download_file: NativeDownloadFile,
73	) -> Digest:
74	return await native_engine.download_file(native_download_file)	12✔
75
76
77	@rule	12✔
78	async def digest_to_snapshot(digest: Digest) -> Snapshot:	12✔
79	return await native_engine.digest_to_snapshot(digest)	12✔
80
81
82	@rule	12✔
83	async def get_digest_contents(digest: Digest) -> DigestContents:	12✔
84	return await native_engine.get_digest_contents(digest)	12✔
85
86
87	@rule	12✔
88	async def get_digest_entries(digest: Digest) -> DigestEntries:	12✔
89	return await native_engine.get_digest_entries(digest)	12✔
90
91
92	@rule	12✔
93	async def merge_digests(merge_digests: MergeDigests) -> Digest:	12✔
94	return await native_engine.merge_digests(merge_digests)	12✔
95
96
97	@rule	12✔
98	async def remove_prefix(remove_prefix: RemovePrefix) -> Digest:	12✔
99	return await native_engine.remove_prefix(remove_prefix)	12✔
100
101
102	@rule	12✔
103	async def add_prefix(add_prefix: AddPrefix) -> Digest:	12✔
104	return await native_engine.add_prefix(add_prefix)	12✔
105
106
107	@rule	12✔
108	async def execute_process(	12✔
109	process: Process, process_execution_environment: ProcessExecutionEnvironment
110	) -> FallibleProcessResult:
111	return await native_engine.execute_process(process, process_execution_environment)	12✔
112
113
114	@rule	12✔
115	async def execute_process_with_retry(req: ProcessWithRetries) -> ProcessResultWithRetries:	12✔
116	results: list[FallibleProcessResult] = []	9✔
117	for attempt in range(0, req.attempts):	9✔
118	proc = dataclasses.replace(req.proc, attempt=attempt)	9✔
119	result = await execute_process(proc, **implicitly())	9✔
120	results.append(result)	9✔
121	if result.exit_code == 0:	9✔
122	break	9✔
123	return ProcessResultWithRetries(tuple(results))	9✔
124
125
126	@rule	12✔
127	async def digest_subset_to_digest(digest_subset: DigestSubset) -> Digest:	12✔
128	return await native_engine.digest_subset_to_digest(digest_subset)	12✔
129
130
131	@rule	12✔
132	async def session_values() -> SessionValues:	12✔
133	return await native_engine.session_values()	12✔
134
135
136	@rule	12✔
137	async def run_id() -> RunId:	12✔
138	return await native_engine.run_id()	1✔
139
140
141	__SQUELCH_WARNING = "__squelch_warning"	12✔
142
143
144	# NB: Call one of the helpers below, instead of calling this rule directly,
145	# to ensure correct application of restartable logic.
146	@_uncacheable_rule	12✔
147	async def _interactive_process(	12✔
148	process: InteractiveProcess, process_execution_environment: ProcessExecutionEnvironment
149	) -> InteractiveProcessResult:
150	# This is a crafty way for a caller to signal into this function without a dedicated arg
151	# (which would confound the solver). Note that we go via __dict__ instead of using
152	# setattr/delattr, because those error for frozen dataclasses.
153	if __SQUELCH_WARNING in process.__dict__:	5✔
154	del process.__dict__[__SQUELCH_WARNING]	5✔
155	else:
156	logging.warning(	×
157	"A plugin is calling `await _interactive_process(...)` directly. This will cause "
158	"restarting logic not to be applied. Use `await run_interactive_process(process)` "
159	"or `await run_interactive_process_in_environment(process, environment_name)` instead. "
160	f"See {git_url('src/python/pants/engine/intrinsics.py')} for more details."
161	)
162	return await native_engine.interactive_process(process, process_execution_environment)	5✔
163
164
165	async def run_interactive_process(process: InteractiveProcess) -> InteractiveProcessResult:	12✔
166	# NB: We must call task_side_effected() in this helper, rather than in a nested @rule call,
167	# so that the Task for the @rule that calls this helper is the one marked as non-restartable.
168	if not process.restartable:	6✔
169	task_side_effected()	6✔
170
171	process.__dict__[__SQUELCH_WARNING] = True	6✔
172	ret: InteractiveProcessResult = await _interactive_process(process, **implicitly())	6✔
173	return ret	6✔
174
175
176	async def run_interactive_process_in_environment(	12✔
177	process: InteractiveProcess, environment_name: EnvironmentName
178	) -> InteractiveProcessResult:
179	# NB: We must call task_side_effected() in this helper, rather than in a nested @rule call,
180	# so that the Task for the @rule that calls this helper is the one marked as non-restartable.
UNCOV 181	if not process.restartable:	1✔
UNCOV 182	task_side_effected()	1✔
183
UNCOV 184	process.__dict__[__SQUELCH_WARNING] = True	1✔
UNCOV 185	ret: InteractiveProcessResult = await _interactive_process(	1✔
186	process, **implicitly({environment_name: EnvironmentName})
187	)
UNCOV 188	return ret	1✔
189
190
191	@rule	12✔
192	async def docker_resolve_image(request: DockerResolveImageRequest) -> DockerResolveImageResult:	12✔
193	return await native_engine.docker_resolve_image(request)	2✔
194
195
196	@rule	12✔
197	async def parse_dockerfile_info(	12✔
198	deps_request: NativeDependenciesRequest,
199	) -> NativeDockerfileInfos:
200	path_infos_pairs = await native_engine.parse_dockerfile_info(deps_request)	7✔
201	return NativeDockerfileInfos(FrozenDict(path_infos_pairs))	7✔
202
203
204	@rule	12✔
205	async def parse_python_deps(	12✔
206	deps_request: NativeDependenciesRequest,
207	) -> NativePythonFilesDependencies:
208	path_deps_pairs = await native_engine.parse_python_deps(deps_request)	11✔
209	return NativePythonFilesDependencies(FrozenDict(path_deps_pairs))	11✔
210
211
212	@rule	12✔
213	async def parse_javascript_deps(	12✔
214	deps_request: NativeDependenciesRequest,
215	) -> NativeJavascriptFilesDependencies:
216	path_deps_pairs = await native_engine.parse_javascript_deps(deps_request)	5✔
217	return NativeJavascriptFilesDependencies(FrozenDict(path_deps_pairs))	5✔
218
219
220	@rule	12✔
221	async def path_metadata_request(request: PathMetadataRequest) -> PathMetadataResult:	12✔
222	return await native_engine.path_metadata_request(request)	12✔
223
224
225	def rules():	12✔
226	return [	12✔
227	*collect_rules(),
228	]

pantsbuild / pants / 25443604553

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