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

96.83

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

# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
# This file should be moved once we figure out where everything belongs

from __future__ import annotations

import logging
from collections.abc import Iterable
from dataclasses import dataclass

from pants.core.target_types import FileSourceField
from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files
from pants.engine.addresses import UnparsedAddressInputs
from pants.engine.fs import DigestSubset, PathGlobs
from pants.engine.internals.graph import resolve_targets
from pants.engine.intrinsics import digest_subset_to_digest, digest_to_snapshot
from pants.engine.rules import Rule, collect_rules, implicitly, rule
from pants.engine.target import (
    COMMON_TARGET_FIELDS,
    GeneratedSources,
    GenerateSourcesRequest,
    MultipleSourcesField,
    SourcesField,
    SpecialCasedDependencies,
    StringSequenceField,
    Target,
)
from pants.engine.unions import UnionRule
from pants.util.strutil import help_text

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class WrapSource:
    rules: tuple[Rule | UnionRule, ...]
    target_types: tuple[type[Target], ...]


class ActivateWrapSourceTargetFieldBase(MultipleSourcesField):
    # We solely register so that codegen can match a fieldset.
    # One unique subclass must be defined per target type.
    alias = "_sources"
    uses_source_roots = False
    expected_num_files = 0


class WrapSourceInputsField(SpecialCasedDependencies):
    alias = "inputs"
    required = True
    help = "The input targets that are to be made available by this target."


class WrapSourceOutputsField(StringSequenceField):
    alias = "outputs"
    required = False
    help = help_text(
        "The output files that are made available in the new context by this target. If not "
        "specified, the target will capture all files with the expected extensions for this "
        "source format: see the help for the target for the specific extensions. If no extensions "
        "are specified and this value is not specified, all input files will be returned."
    )


async def _wrap_source(wrapper: GenerateSourcesRequest) -> GeneratedSources:
    request = wrapper.protocol_target
    default_extensions = {i for i in (wrapper.output.expected_file_extensions or ()) if i}

    inputs = await resolve_targets(
        **implicitly(
            {request.get(WrapSourceInputsField).to_unparsed_address_inputs(): UnparsedAddressInputs}
        )
    )

    sources = await determine_source_files(
        SourceFilesRequest(
            sources_fields=[tgt.get(SourcesField) for tgt in inputs],
            for_sources_types=(SourcesField, FileSourceField),
            enable_codegen=True,
        ),
    )

    outputs_value: Iterable[str] | None = request.get(WrapSourceOutputsField).value
    if outputs_value:
        pass
    elif default_extensions:
        outputs_value = [i for i in sources.files if any(i.endswith(j) for j in default_extensions)]
    else:
        outputs_value = sources.files

    filter_digest = await digest_subset_to_digest(
        DigestSubset(sources.snapshot.digest, PathGlobs(outputs_value))
    )

    snapshot = await digest_to_snapshot(filter_digest)
    return GeneratedSources(snapshot)


def wrap_source_rule_and_target(
    source_field_type: type[SourcesField], target_name_suffix: str
) -> WrapSource:
    if source_field_type.expected_file_extensions:
        outputs_help = (
            "If `outputs` is not specified, all files with the following extensions will be "
            "matched: "
            + ", ".join(ext for ext in source_field_type.expected_file_extensions if ext)
        )
    else:
        outputs_help = "If `outputs` is not specified, all files from `inputs` will be matched."

    class ActivateWrapSourceTargetField(ActivateWrapSourceTargetFieldBase):
        pass

    class GenerateWrapSourceSourcesRequest(GenerateSourcesRequest):
        input = ActivateWrapSourceTargetField
        output = source_field_type

    setattr(
        GenerateWrapSourceSourcesRequest,
        "__qualname__",
        f"{source_field_type.__qualname__}.{GenerateWrapSourceSourcesRequest.__name__}",
    )

    class WrapSourceTarget(Target):
        alias = f"experimental_wrap_as_{target_name_suffix}"
        core_fields = (
            *COMMON_TARGET_FIELDS,
            ActivateWrapSourceTargetField,
            WrapSourceInputsField,
            WrapSourceOutputsField,
        )
        help = help_text(
            f"""
            Allow files and sources produced by the targets specified by `inputs` to be consumed
            by rules that specifically expect a `{source_field_type.__name__}`.

            Note that this target does not modify the files in any way. {outputs_help}

            This target must be explicitly specified as a dependency of any target that requires
            it. Sources from this target will not be automatically inferred as dependencies.

            This target is experimental: in future versions of Pants, this functionality may be
            made available with a different interface.
            """
        )

    # need to use `_param_type_overrides` to stop `@rule` from inspecting the function's source
    @rule(
        canonical_name_suffix=source_field_type.__name__,
        _param_type_overrides={"request": GenerateWrapSourceSourcesRequest},
    )
    async def wrap_source(request: GenerateSourcesRequest) -> GeneratedSources:
        return await _wrap_source(request)

    return WrapSource(
        rules=(
            *collect_rules(locals()),
            UnionRule(GenerateSourcesRequest, GenerateWrapSourceSourcesRequest),
        ),
        target_types=(WrapSourceTarget,),
    )

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	# This file should be moved once we figure out where everything belongs
4
5	from __future__ import annotations	9✔
6
7	import logging	9✔
8	from collections.abc import Iterable	9✔
9	from dataclasses import dataclass	9✔
10
11	from pants.core.target_types import FileSourceField	9✔
12	from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files	9✔
13	from pants.engine.addresses import UnparsedAddressInputs	9✔
14	from pants.engine.fs import DigestSubset, PathGlobs	9✔
15	from pants.engine.internals.graph import resolve_targets	9✔
16	from pants.engine.intrinsics import digest_subset_to_digest, digest_to_snapshot	9✔
17	from pants.engine.rules import Rule, collect_rules, implicitly, rule	9✔
18	from pants.engine.target import (	9✔
19	COMMON_TARGET_FIELDS,
20	GeneratedSources,
21	GenerateSourcesRequest,
22	MultipleSourcesField,
23	SourcesField,
24	SpecialCasedDependencies,
25	StringSequenceField,
26	Target,
27	)
28	from pants.engine.unions import UnionRule	9✔
29	from pants.util.strutil import help_text	9✔
30
31	logger = logging.getLogger(__name__)	9✔
32
33
34	@dataclass(frozen=True)	9✔
35	class WrapSource:	9✔
36	rules: tuple[Rule \| UnionRule, ...]	9✔
37	target_types: tuple[type[Target], ...]	9✔
38
39
40	class ActivateWrapSourceTargetFieldBase(MultipleSourcesField):	9✔
41	# We solely register so that codegen can match a fieldset.
42	# One unique subclass must be defined per target type.
43	alias = "_sources"	9✔
44	uses_source_roots = False	9✔
45	expected_num_files = 0	9✔
46
47
48	class WrapSourceInputsField(SpecialCasedDependencies):	9✔
49	alias = "inputs"	9✔
50	required = True	9✔
51	help = "The input targets that are to be made available by this target."	9✔
52
53
54	class WrapSourceOutputsField(StringSequenceField):	9✔
55	alias = "outputs"	9✔
56	required = False	9✔
57	help = help_text(	9✔
58	"The output files that are made available in the new context by this target. If not "
59	"specified, the target will capture all files with the expected extensions for this "
60	"source format: see the help for the target for the specific extensions. If no extensions "
61	"are specified and this value is not specified, all input files will be returned."
62	)
63
64
65	async def _wrap_source(wrapper: GenerateSourcesRequest) -> GeneratedSources:	9✔
UNCOV 66	request = wrapper.protocol_target	1✔
UNCOV 67	default_extensions = {i for i in (wrapper.output.expected_file_extensions or ()) if i}	1✔
68
UNCOV 69	inputs = await resolve_targets(	1✔
70	**implicitly(
71	{request.get(WrapSourceInputsField).to_unparsed_address_inputs(): UnparsedAddressInputs}
72	)
73	)
74
UNCOV 75	sources = await determine_source_files(	1✔
76	SourceFilesRequest(
77	sources_fields=[tgt.get(SourcesField) for tgt in inputs],
78	for_sources_types=(SourcesField, FileSourceField),
79	enable_codegen=True,
80	),
81	)
82
UNCOV 83	outputs_value: Iterable[str] \| None = request.get(WrapSourceOutputsField).value	1✔
UNCOV 84	if outputs_value:	1✔
85	pass	×
UNCOV 86	elif default_extensions:	1✔
87	outputs_value = [i for i in sources.files if any(i.endswith(j) for j in default_extensions)]	×
88	else:
UNCOV 89	outputs_value = sources.files	1✔
90
UNCOV 91	filter_digest = await digest_subset_to_digest(	1✔
92	DigestSubset(sources.snapshot.digest, PathGlobs(outputs_value))
93	)
94
UNCOV 95	snapshot = await digest_to_snapshot(filter_digest)	1✔
UNCOV 96	return GeneratedSources(snapshot)	1✔
97
98
99	def wrap_source_rule_and_target(	9✔
100	source_field_type: type[SourcesField], target_name_suffix: str
101	) -> WrapSource:
102	if source_field_type.expected_file_extensions:	9✔
103	outputs_help = (	6✔
104	"If `outputs` is not specified, all files with the following extensions will be "
105	"matched: "
106	+ ", ".join(ext for ext in source_field_type.expected_file_extensions if ext)
107	)
108	else:
109	outputs_help = "If `outputs` is not specified, all files from `inputs` will be matched."	7✔
110
111	class ActivateWrapSourceTargetField(ActivateWrapSourceTargetFieldBase):	9✔
112	pass	9✔
113
114	class GenerateWrapSourceSourcesRequest(GenerateSourcesRequest):	9✔
115	input = ActivateWrapSourceTargetField	9✔
116	output = source_field_type	9✔
117
118	setattr(	9✔
119	GenerateWrapSourceSourcesRequest,
120	"__qualname__",
121	f"{source_field_type.__qualname__}.{GenerateWrapSourceSourcesRequest.__name__}",
122	)
123
124	class WrapSourceTarget(Target):	9✔
125	alias = f"experimental_wrap_as_{target_name_suffix}"	9✔
126	core_fields = (	9✔
127	*COMMON_TARGET_FIELDS,
128	ActivateWrapSourceTargetField,
129	WrapSourceInputsField,
130	WrapSourceOutputsField,
131	)
132	help = help_text(	9✔
133	f"""
134	Allow files and sources produced by the targets specified by `inputs` to be consumed
135	by rules that specifically expect a `{source_field_type.__name__}`.
136
137	Note that this target does not modify the files in any way. {outputs_help}
138
139	This target must be explicitly specified as a dependency of any target that requires
140	it. Sources from this target will not be automatically inferred as dependencies.
141
142	This target is experimental: in future versions of Pants, this functionality may be
143	made available with a different interface.
144	"""
145	)
146
147	# need to use `_param_type_overrides` to stop `@rule` from inspecting the function's source
148	@rule(	9✔
149	canonical_name_suffix=source_field_type.__name__,
150	_param_type_overrides={"request": GenerateWrapSourceSourcesRequest},
151	)
152	async def wrap_source(request: GenerateSourcesRequest) -> GeneratedSources:	9✔
UNCOV 153	return await _wrap_source(request)	1✔
154
155	return WrapSource(	9✔
156	rules=(
157	*collect_rules(locals()),
158	UnionRule(GenerateSourcesRequest, GenerateWrapSourceSourcesRequest),
159	),
160	target_types=(WrapSourceTarget,),
161	)

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