25441711719

Committed 06 May 2026 02:31PM UTC coverage: 92.915%. Remained the same

Build # 25441711719

Build Type

push

github

Committed by

web-flow

Commit Message

use sha pin (with comment) format for generated actions (#23312)

Per the GitHub Action best practices we recently enabled at #23249, we
should pin each action to a SHA so that the reference is actually
immutable.

This will -- I hope -- knock out a large chunk of the 421 alerts we
currently get from zizmor. The next followup would then be upgrades and
harmonizing the generated and none-generated pins.

Notice: This idea was suggested by Claude while going over pinact output
and I was surprised to see that post processing the yaml wasn't too
gross.

Coverage Stats

92206 of 99237 relevant lines covered (92.91%)

4.04 hits per line

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

98.81

/src/python/pants/backend/codegen/protobuf/python/rules.py

# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
import logging
import os
from pathlib import PurePath

from pants.backend.codegen.protobuf import protoc
from pants.backend.codegen.protobuf.protoc import Protoc
from pants.backend.codegen.protobuf.python.additional_fields import PythonSourceRootField
from pants.backend.codegen.protobuf.python.grpc_python_plugin import GrpcPythonPlugin
from pants.backend.codegen.protobuf.python.python_protobuf_subsystem import (
    PythonProtobufGrpclibPlugin,
    PythonProtobufMypyPlugin,
    PythonProtobufSubsystem,
)
from pants.backend.codegen.protobuf.target_types import ProtobufGrpcToggleField, ProtobufSourceField
from pants.backend.python.target_types import PythonSourceField
from pants.backend.python.util_rules import pex
from pants.backend.python.util_rules.pex import (
    VenvPexRequest,
    create_venv_pex,
    determine_venv_pex_resolve_info,
)
from pants.backend.python.util_rules.pex_environment import PexEnvironment
from pants.core.goals.resolves import ExportableTool
from pants.core.util_rules.external_tool import download_external_tool
from pants.core.util_rules.source_files import SourceFilesRequest
from pants.core.util_rules.stripped_source_files import strip_source_roots
from pants.engine.fs import AddPrefix, CreateDigest, Directory, MergeDigests, RemovePrefix
from pants.engine.internals.graph import transitive_targets as transitive_targets_get
from pants.engine.intrinsics import create_digest, digest_to_snapshot, merge_digests, remove_prefix
from pants.engine.platform import Platform
from pants.engine.process import Process, execute_process_or_raise
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import GeneratedSources, GenerateSourcesRequest, TransitiveTargetsRequest
from pants.engine.unions import UnionRule
from pants.source.source_root import SourceRootRequest, get_source_root
from pants.util.logging import LogLevel

logger = logging.getLogger(__name__)


class GeneratePythonFromProtobufRequest(GenerateSourcesRequest):
    input = ProtobufSourceField
    output = PythonSourceField


@rule(desc="Generate Python from Protobuf", level=LogLevel.DEBUG)
async def generate_python_from_protobuf(
    request: GeneratePythonFromProtobufRequest,
    protoc: Protoc,
    grpc_python_plugin: GrpcPythonPlugin,
    python_protobuf_subsystem: PythonProtobufSubsystem,
    python_protobuf_mypy_plugin: PythonProtobufMypyPlugin,
    python_protobuf_grpclib_plugin: PythonProtobufGrpclibPlugin,
    pex_environment: PexEnvironment,
    platform: Platform,
) -> GeneratedSources:
    download_protoc_request = download_external_tool(protoc.get_request(platform))

    output_dir = "_generated_files"
    create_output_dir_request = create_digest(CreateDigest([Directory(output_dir)]))

    # Protoc needs all transitive dependencies on `protobuf_libraries` to work properly. It won't
    # actually generate those dependencies; it only needs to look at their .proto files to work
    # with imports.
    transitive_targets = await transitive_targets_get(
        TransitiveTargetsRequest([request.protocol_target.address]), **implicitly()
    )

    # NB: By stripping the source roots, we avoid having to set the value `--proto_path`
    # for Protobuf imports to be discoverable.
    all_stripped_sources_request = strip_source_roots(
        **implicitly(
            SourceFilesRequest(
                tgt[ProtobufSourceField]
                for tgt in transitive_targets.closure
                if tgt.has_field(ProtobufSourceField)
            )
        )
    )
    target_stripped_sources_request = strip_source_roots(
        **implicitly(SourceFilesRequest([request.protocol_target[ProtobufSourceField]]))
    )

    (
        downloaded_protoc_binary,
        empty_output_dir,
        all_sources_stripped,
        target_sources_stripped,
    ) = await concurrently(
        download_protoc_request,
        create_output_dir_request,
        all_stripped_sources_request,
        target_stripped_sources_request,
    )

    grpc_enabled = request.protocol_target.get(ProtobufGrpcToggleField).value
    protoc_relpath = "__protoc"
    unmerged_digests = [
        all_sources_stripped.snapshot.digest,
        empty_output_dir,
    ]

    pyi_gen_option = "pyi_out:" if python_protobuf_subsystem.generate_type_stubs else ""
    protoc_argv = [
        os.path.join(protoc_relpath, downloaded_protoc_binary.exe),
        f"--python_out={pyi_gen_option}{output_dir}",
    ]

    complete_pex_env = pex_environment.in_sandbox(working_directory=None)

    if python_protobuf_subsystem.mypy_plugin:
        protoc_gen_mypy_script = "protoc-gen-mypy"
        protoc_gen_mypy_grpc_script = "protoc-gen-mypy_grpc"
        mypy_request = python_protobuf_mypy_plugin.to_pex_request()
        mypy_pex = await create_venv_pex(
            VenvPexRequest(
                pex_request=mypy_request,
                complete_pex_env=complete_pex_env,
                bin_names=[protoc_gen_mypy_script],
            ),
            **implicitly(),
        )
        protoc_argv.extend(
            [
                f"--plugin=protoc-gen-mypy={mypy_pex.bin[protoc_gen_mypy_script].argv0}",
                "--mypy_out",
                output_dir,
            ]
        )

        if grpc_enabled and python_protobuf_subsystem.grpcio_plugin:
            mypy_pex_info = await determine_venv_pex_resolve_info(mypy_pex)

            # In order to generate stubs for gRPC code, we need mypy-protobuf 2.0 or above.
            mypy_protobuf_info = mypy_pex_info.find("mypy-protobuf")
            if mypy_protobuf_info and mypy_protobuf_info.version.major >= 2:
                # TODO: Use `pex_path` once VenvPex stores a Pex field.
                mypy_pex = await create_venv_pex(
                    VenvPexRequest(
                        pex_request=mypy_request,
                        complete_pex_env=complete_pex_env,
                        bin_names=[protoc_gen_mypy_script, protoc_gen_mypy_grpc_script],
                    ),
                    **implicitly(),
                )
                protoc_argv.extend(
                    [
                        f"--plugin=protoc-gen-mypy_grpc={mypy_pex.bin[protoc_gen_mypy_grpc_script].argv0}",
                        "--mypy_grpc_out",
                        output_dir,
                    ]
                )
        unmerged_digests.append(mypy_pex.digest)

    if grpc_enabled:
        if not (
            python_protobuf_subsystem.grpcio_plugin or python_protobuf_subsystem.grpclib_plugin
        ):
            logger.warning(
                """
            No Python grpc plugins have been enabled. Make sure to enable at least one of the
            following under the [python-protobuf] configuration: grpcio_plugin, grpclib_plugin.
            """
            )

        if python_protobuf_subsystem.grpcio_plugin:
            downloaded_grpc_plugin = await download_external_tool(
                grpc_python_plugin.get_request(platform)
            )
            unmerged_digests.append(downloaded_grpc_plugin.digest)
            protoc_argv.extend(
                [f"--plugin=protoc-gen-grpc={downloaded_grpc_plugin.exe}", "--grpc_out", output_dir]
            )

        if python_protobuf_subsystem.grpclib_plugin:
            protoc_gen_grpclib_script = "protoc-gen-grpclib_python"
            grpclib_request = python_protobuf_grpclib_plugin.to_pex_request()
            grpclib_pex = await create_venv_pex(
                VenvPexRequest(
                    pex_request=grpclib_request,
                    complete_pex_env=complete_pex_env,
                    bin_names=[protoc_gen_grpclib_script],
                ),
                **implicitly(),
            )
            unmerged_digests.append(grpclib_pex.digest)
            protoc_argv.extend(
                [
                    f"--plugin=protoc-gen-grpclib_python={grpclib_pex.bin[protoc_gen_grpclib_script].argv0}",
                    "--grpclib_python_out",
                    output_dir,
                ]
            )

    input_digest = await merge_digests(MergeDigests(unmerged_digests))
    protoc_argv.extend(target_sources_stripped.snapshot.files)
    result = await execute_process_or_raise(
        **implicitly(
            Process(
                protoc_argv,
                input_digest=input_digest,
                immutable_input_digests={
                    protoc_relpath: downloaded_protoc_binary.digest,
                },
                description=f"Generating Python sources from {request.protocol_target.address}.",
                level=LogLevel.DEBUG,
                output_directories=(output_dir,),
                append_only_caches=complete_pex_env.append_only_caches,
            )
        ),
    )

    # We must do some path manipulation on the output digest for it to look like normal sources,
    # including adding back a source root.
    py_source_root = request.protocol_target.get(PythonSourceRootField).value
    if py_source_root:
        # Verify that the python source root specified by the target is in fact a source root.
        source_root_request = SourceRootRequest(PurePath(py_source_root))
    else:
        # The target didn't specify a python source root, so use the protobuf_source's source root.
        source_root_request = SourceRootRequest.for_target(request.protocol_target)

    normalized_digest, source_root = await concurrently(
        remove_prefix(RemovePrefix(result.output_digest, output_dir)),
        get_source_root(source_root_request),
    )

    source_root_restored = (
        await digest_to_snapshot(**implicitly(AddPrefix(normalized_digest, source_root.path)))
        if source_root.path != "."
        else await digest_to_snapshot(normalized_digest)
    )
    return GeneratedSources(source_root_restored)


def rules():
    return [
        *collect_rules(),
        *pex.rules(),
        UnionRule(GenerateSourcesRequest, GeneratePythonFromProtobufRequest),
        *protoc.rules(),
        UnionRule(ExportableTool, GrpcPythonPlugin),
    ]

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	import logging	6✔
4	import os	6✔
5	from pathlib import PurePath	6✔
6
7	from pants.backend.codegen.protobuf import protoc	6✔
8	from pants.backend.codegen.protobuf.protoc import Protoc	6✔
9	from pants.backend.codegen.protobuf.python.additional_fields import PythonSourceRootField	6✔
10	from pants.backend.codegen.protobuf.python.grpc_python_plugin import GrpcPythonPlugin	6✔
11	from pants.backend.codegen.protobuf.python.python_protobuf_subsystem import (	6✔
12	PythonProtobufGrpclibPlugin,
13	PythonProtobufMypyPlugin,
14	PythonProtobufSubsystem,
15	)
16	from pants.backend.codegen.protobuf.target_types import ProtobufGrpcToggleField, ProtobufSourceField	6✔
17	from pants.backend.python.target_types import PythonSourceField	6✔
18	from pants.backend.python.util_rules import pex	6✔
19	from pants.backend.python.util_rules.pex import (	6✔
20	VenvPexRequest,
21	create_venv_pex,
22	determine_venv_pex_resolve_info,
23	)
24	from pants.backend.python.util_rules.pex_environment import PexEnvironment	6✔
25	from pants.core.goals.resolves import ExportableTool	6✔
26	from pants.core.util_rules.external_tool import download_external_tool	6✔
27	from pants.core.util_rules.source_files import SourceFilesRequest	6✔
28	from pants.core.util_rules.stripped_source_files import strip_source_roots	6✔
29	from pants.engine.fs import AddPrefix, CreateDigest, Directory, MergeDigests, RemovePrefix	6✔
30	from pants.engine.internals.graph import transitive_targets as transitive_targets_get	6✔
31	from pants.engine.intrinsics import create_digest, digest_to_snapshot, merge_digests, remove_prefix	6✔
32	from pants.engine.platform import Platform	6✔
33	from pants.engine.process import Process, execute_process_or_raise	6✔
34	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	6✔
35	from pants.engine.target import GeneratedSources, GenerateSourcesRequest, TransitiveTargetsRequest	6✔
36	from pants.engine.unions import UnionRule	6✔
37	from pants.source.source_root import SourceRootRequest, get_source_root	6✔
38	from pants.util.logging import LogLevel	6✔
39
40	logger = logging.getLogger(__name__)	6✔
41
42
43	class GeneratePythonFromProtobufRequest(GenerateSourcesRequest):	6✔
44	input = ProtobufSourceField	6✔
45	output = PythonSourceField	6✔
46
47
48	@rule(desc="Generate Python from Protobuf", level=LogLevel.DEBUG)	6✔
49	async def generate_python_from_protobuf(	6✔
50	request: GeneratePythonFromProtobufRequest,
51	protoc: Protoc,
52	grpc_python_plugin: GrpcPythonPlugin,
53	python_protobuf_subsystem: PythonProtobufSubsystem,
54	python_protobuf_mypy_plugin: PythonProtobufMypyPlugin,
55	python_protobuf_grpclib_plugin: PythonProtobufGrpclibPlugin,
56	pex_environment: PexEnvironment,
57	platform: Platform,
58	) -> GeneratedSources:
59	download_protoc_request = download_external_tool(protoc.get_request(platform))	6✔
60
61	output_dir = "_generated_files"	6✔
62	create_output_dir_request = create_digest(CreateDigest([Directory(output_dir)]))	6✔
63
64	# Protoc needs all transitive dependencies on `protobuf_libraries` to work properly. It won't
65	# actually generate those dependencies; it only needs to look at their .proto files to work
66	# with imports.
67	transitive_targets = await transitive_targets_get(	6✔
68	TransitiveTargetsRequest([request.protocol_target.address]), **implicitly()
69	)
70
71	# NB: By stripping the source roots, we avoid having to set the value `--proto_path`
72	# for Protobuf imports to be discoverable.
73	all_stripped_sources_request = strip_source_roots(	6✔
74	**implicitly(
75	SourceFilesRequest(
76	tgt[ProtobufSourceField]
77	for tgt in transitive_targets.closure
78	if tgt.has_field(ProtobufSourceField)
79	)
80	)
81	)
82	target_stripped_sources_request = strip_source_roots(	6✔
83	**implicitly(SourceFilesRequest([request.protocol_target[ProtobufSourceField]]))
84	)
85
86	(	6✔
87	downloaded_protoc_binary,
88	empty_output_dir,
89	all_sources_stripped,
90	target_sources_stripped,
91	) = await concurrently(
92	download_protoc_request,
93	create_output_dir_request,
94	all_stripped_sources_request,
95	target_stripped_sources_request,
96	)
97
98	grpc_enabled = request.protocol_target.get(ProtobufGrpcToggleField).value	6✔
99	protoc_relpath = "__protoc"	6✔
100	unmerged_digests = [	6✔
101	all_sources_stripped.snapshot.digest,
102	empty_output_dir,
103	]
104
105	pyi_gen_option = "pyi_out:" if python_protobuf_subsystem.generate_type_stubs else ""	6✔
106	protoc_argv = [	6✔
107	os.path.join(protoc_relpath, downloaded_protoc_binary.exe),
108	f"--python_out={pyi_gen_option}{output_dir}",
109	]
110
111	complete_pex_env = pex_environment.in_sandbox(working_directory=None)	6✔
112
113	if python_protobuf_subsystem.mypy_plugin:	6✔
114	protoc_gen_mypy_script = "protoc-gen-mypy"	4✔
115	protoc_gen_mypy_grpc_script = "protoc-gen-mypy_grpc"	4✔
116	mypy_request = python_protobuf_mypy_plugin.to_pex_request()	4✔
117	mypy_pex = await create_venv_pex(	4✔
118	VenvPexRequest(
119	pex_request=mypy_request,
120	complete_pex_env=complete_pex_env,
121	bin_names=[protoc_gen_mypy_script],
122	),
123	**implicitly(),
124	)
125	protoc_argv.extend(	4✔
126	[
127	f"--plugin=protoc-gen-mypy={mypy_pex.bin[protoc_gen_mypy_script].argv0}",
128	"--mypy_out",
129	output_dir,
130	]
131	)
132
133	if grpc_enabled and python_protobuf_subsystem.grpcio_plugin:	4✔
134	mypy_pex_info = await determine_venv_pex_resolve_info(mypy_pex)	3✔
135
136	# In order to generate stubs for gRPC code, we need mypy-protobuf 2.0 or above.
137	mypy_protobuf_info = mypy_pex_info.find("mypy-protobuf")	3✔
138	if mypy_protobuf_info and mypy_protobuf_info.version.major >= 2:	3✔
139	# TODO: Use `pex_path` once VenvPex stores a Pex field.
140	mypy_pex = await create_venv_pex(	3✔
141	VenvPexRequest(
142	pex_request=mypy_request,
143	complete_pex_env=complete_pex_env,
144	bin_names=[protoc_gen_mypy_script, protoc_gen_mypy_grpc_script],
145	),
146	**implicitly(),
147	)
148	protoc_argv.extend(	3✔
149	[
150	f"--plugin=protoc-gen-mypy_grpc={mypy_pex.bin[protoc_gen_mypy_grpc_script].argv0}",
151	"--mypy_grpc_out",
152	output_dir,
153	]
154	)
155	unmerged_digests.append(mypy_pex.digest)	4✔
156
157	if grpc_enabled:	6✔
158	if not (	3✔
159	python_protobuf_subsystem.grpcio_plugin or python_protobuf_subsystem.grpclib_plugin
160	):
161	logger.warning(	×
162	"""
163	No Python grpc plugins have been enabled. Make sure to enable at least one of the
164	following under the [python-protobuf] configuration: grpcio_plugin, grpclib_plugin.
165	"""
166	)
167
168	if python_protobuf_subsystem.grpcio_plugin:	3✔
169	downloaded_grpc_plugin = await download_external_tool(	3✔
170	grpc_python_plugin.get_request(platform)
171	)
172	unmerged_digests.append(downloaded_grpc_plugin.digest)	3✔
173	protoc_argv.extend(	3✔
174	[f"--plugin=protoc-gen-grpc={downloaded_grpc_plugin.exe}", "--grpc_out", output_dir]
175	)
176
177	if python_protobuf_subsystem.grpclib_plugin:	3✔
178	protoc_gen_grpclib_script = "protoc-gen-grpclib_python"	3✔
179	grpclib_request = python_protobuf_grpclib_plugin.to_pex_request()	3✔
180	grpclib_pex = await create_venv_pex(	3✔
181	VenvPexRequest(
182	pex_request=grpclib_request,
183	complete_pex_env=complete_pex_env,
184	bin_names=[protoc_gen_grpclib_script],
185	),
186	**implicitly(),
187	)
188	unmerged_digests.append(grpclib_pex.digest)	3✔
189	protoc_argv.extend(	3✔
190	[
191	f"--plugin=protoc-gen-grpclib_python={grpclib_pex.bin[protoc_gen_grpclib_script].argv0}",
192	"--grpclib_python_out",
193	output_dir,
194	]
195	)
196
197	input_digest = await merge_digests(MergeDigests(unmerged_digests))	6✔
198	protoc_argv.extend(target_sources_stripped.snapshot.files)	6✔
199	result = await execute_process_or_raise(	6✔
200	**implicitly(
201	Process(
202	protoc_argv,
203	input_digest=input_digest,
204	immutable_input_digests={
205	protoc_relpath: downloaded_protoc_binary.digest,
206	},
207	description=f"Generating Python sources from {request.protocol_target.address}.",
208	level=LogLevel.DEBUG,
209	output_directories=(output_dir,),
210	append_only_caches=complete_pex_env.append_only_caches,
211	)
212	),
213	)
214
215	# We must do some path manipulation on the output digest for it to look like normal sources,
216	# including adding back a source root.
217	py_source_root = request.protocol_target.get(PythonSourceRootField).value	6✔
218	if py_source_root:	6✔
219	# Verify that the python source root specified by the target is in fact a source root.
220	source_root_request = SourceRootRequest(PurePath(py_source_root))	2✔
221	else:
222	# The target didn't specify a python source root, so use the protobuf_source's source root.
223	source_root_request = SourceRootRequest.for_target(request.protocol_target)	6✔
224
225	normalized_digest, source_root = await concurrently(	6✔
226	remove_prefix(RemovePrefix(result.output_digest, output_dir)),
227	get_source_root(source_root_request),
228	)
229
230	source_root_restored = (	6✔
231	await digest_to_snapshot(**implicitly(AddPrefix(normalized_digest, source_root.path)))
232	if source_root.path != "."
233	else await digest_to_snapshot(normalized_digest)
234	)
235	return GeneratedSources(source_root_restored)	6✔
236
237
238	def rules():	6✔
239	return [	6✔
240	*collect_rules(),
241	*pex.rules(),
242	UnionRule(GenerateSourcesRequest, GeneratePythonFromProtobufRequest),
243	*protoc.rules(),
244	UnionRule(ExportableTool, GrpcPythonPlugin),
245	]

pantsbuild / pants / 25441711719

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