26689585807

Committed 30 May 2026 04:55PM UTC coverage: 92.742% (-0.05%) from 92.792%

Build # 26689585807

Build Type

Pull #23343

github

Committed by

web-flow

Commit Message

Merge c42efe377 into c8127c1f4

Pull Request Pull Request #23343: Add buf as an alternate Python protobuf code generator

Coverage Stats

767 of 807 new or added lines in 17 files covered. (95.04%)

69 existing lines in 3 files now uncovered.

93753 of 101090 relevant lines covered (92.74%)

4.01 hits per line

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

98.48

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

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

from __future__ import annotations

import logging
import os
from collections import defaultdict
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import DefaultDict

from pants.backend.codegen.protobuf.buf.config import (
    BufGenContent,
    BufLayout,
    fetch_buf_gen_contents,
    fetch_buf_layout,
    parse_plugin_outs,
    suffix_plugin_includes_imports,
)
from pants.backend.codegen.protobuf.buf.subsystem import BufSubsystem
from pants.backend.codegen.protobuf.python.additional_fields import PythonSourceRootField
from pants.backend.codegen.protobuf.python.python_protobuf_subsystem import (
    DEFAULT_BSR_DEP_MODULES,
    DEFAULT_PLUGIN_SUFFIXES,
    PythonProtobufSubsystem,
)
from pants.backend.codegen.protobuf.target_types import (
    AllProtobufTargets,
    ProtobufGeneratorField,
    ProtobufGrpcToggleField,
    ProtobufSourceField,
)
from pants.backend.python.dependency_inference.module_mapper import (
    FirstPartyPythonMappingImpl,
    FirstPartyPythonMappingImplMarker,
    ModuleProvider,
    ModuleProviderType,
    ResolveName,
)
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.target_types import PythonResolveField
from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name
from pants.engine.rules import collect_rules, concurrently, rule
from pants.engine.target import Target
from pants.engine.unions import UnionRule
from pants.util.logging import LogLevel

logger = logging.getLogger(__name__)


def proto_path_to_py_module(stripped_path: str, *, suffix: str) -> str:
    return stripped_path.replace(".proto", suffix).replace("/", ".")


# This is only used to register our implementation with the plugin hook via unions.
class PythonProtobufMappingMarker(FirstPartyPythonMappingImplMarker):
    pass


# Suffixes relevant to Python codegen. Each is registered iff its plugin appears
# in `buf.gen.yaml`.
_PB2_SUFFIX = "_pb2"
_SERVICE_SUFFIXES: tuple[str, ...] = ("_pb2_grpc", "_grpc", "_connect")


@dataclass(frozen=True)
class _BufStripPlan:
    """Plan for one buf target: paths to feed to `strip_file_name`, paired with the
    suffix to apply to each stripped result."""

    suffixes: tuple[str, ...]
    paths_to_strip: tuple[str, ...]


def _plan_buf_target(
    target: Target,
    suffix_outs: Mapping[str, str],
    buf_module_root: str,
) -> _BufStripPlan:
    """Build the strip plan from the suffixes matched in this target's `buf.gen.yaml`.

    A suffix is registered iff its plugin appears in the file. `grpc=True` is
    *not* consulted: `buf.gen.yaml` is the authoritative source of truth for
    which buf target outputs exist.
    """
    proto_path = target[ProtobufSourceField].file_path
    rel_proto = (
        os.path.relpath(proto_path, buf_module_root)
        if buf_module_root
        and (proto_path == buf_module_root or proto_path.startswith(buf_module_root + os.sep))
        else proto_path
    )

    def _path_for(out_dir: str) -> str:
        return os.path.normpath(os.path.join(out_dir, rel_proto))

    suffixes: list[str] = []
    paths: list[str] = []
    if _PB2_SUFFIX in suffix_outs:
        suffixes.append(_PB2_SUFFIX)
        paths.append(_path_for(suffix_outs[_PB2_SUFFIX]))
    for suffix in _SERVICE_SUFFIXES:
        if suffix in suffix_outs:
            suffixes.append(suffix)
            paths.append(_path_for(suffix_outs[suffix]))

    return _BufStripPlan(tuple(suffixes), tuple(paths))


def _fallback_plan(target: Target) -> _BufStripPlan:
    """Plan when no `buf.gen.yaml` was found: assume the proto's source root also
    covers the generated `.py`. Service suffixes can't be inferred without the
    template, so we register only `_pb2`."""
    proto_path = target[ProtobufSourceField].file_path
    return _BufStripPlan((_PB2_SUFFIX,), (proto_path,))


# Protoc-only subsystem options. Their default values are mirrored here so we can
# detect when the user explicitly set them while also using buf targets, and warn
# that they're ignored on the buf path. Keep in sync with the option definitions
# in `python_protobuf_subsystem.py`.
_PROTOC_ONLY_OPTION_DEFAULTS: tuple[tuple[str, object], ...] = (
    ("grpcio_plugin", True),
    ("grpclib_plugin", False),
    ("mypy_plugin", False),
    ("generate_type_stubs", False),
)


def _emit_subsystem_warnings_for_buf(subsystem: PythonProtobufSubsystem) -> None:
    """Warn once if subsystem options that are protoc-only are set non-default
    while at least one buf target exists."""
    for option_name, default in _PROTOC_ONLY_OPTION_DEFAULTS:
        if getattr(subsystem, option_name) == default:
            continue
        logger.warning(
            "[%s].%s is set but ignored for `protobuf_generator='buf'` targets. "
            "Service generation and `.pyi` stubs for buf targets are determined by "
            "the plugin entries in `buf.gen.yaml`.",
            subsystem.options_scope,
            option_name,
        )


def _emit_per_target_warnings_for_buf(target: Target) -> None:
    """Warn once per buf target about field values that have no effect."""
    if target.get(ProtobufGrpcToggleField).value:
        logger.warning(
            "`grpc=True` is set on %s but is ignored for `protobuf_generator='buf'` "
            "targets. Whether `_pb2_grpc.py` / `_grpc.py` / `_connect.py` exist is "
            "determined by the plugins in `buf.gen.yaml`.",
            target.address,
        )
    if target.get(PythonSourceRootField).value is not None:
        logger.warning(
            "`python_source_root` is set on %s but ignored for "
            "`protobuf_generator='buf'`; output paths come from the `out:` field of "
            "`buf.gen.yaml`.",
            target.address,
        )


@rule(desc="Creating map of Protobuf targets to generated Python modules", level=LogLevel.DEBUG)
async def map_protobuf_to_python_modules(
    protobuf_targets: AllProtobufTargets,
    python_setup: PythonSetup,
    python_protobuf_subsystem: PythonProtobufSubsystem,
    buf: BufSubsystem,
    _: PythonProtobufMappingMarker,
) -> FirstPartyPythonMappingImpl:
    grpc_suffixes_list: list[str] = []
    if python_protobuf_subsystem.grpcio_plugin:
        grpc_suffixes_list.append("_pb2_grpc")
    if python_protobuf_subsystem.grpclib_plugin:
        grpc_suffixes_list.append("_grpc")
    grpc_suffixes = tuple(grpc_suffixes_list)

    protoc_targets: list[Target] = []
    buf_targets: list[Target] = []
    for tgt in protobuf_targets:
        if tgt.get(ProtobufGeneratorField).value == "buf":
            buf_targets.append(tgt)
        else:
            protoc_targets.append(tgt)

    if buf_targets:
        _emit_subsystem_warnings_for_buf(python_protobuf_subsystem)

    # ---- protoc path. ----
    stripped_file_per_protoc_target = await concurrently(
        strip_file_name(StrippedFileNameRequest(tgt[ProtobufSourceField].file_path))
        for tgt in protoc_targets
    )

    # ---- buf path: registry-driven plugin matching, no `grpc=True` gate. ----
    if buf_targets:
        buf_layout: BufLayout = await fetch_buf_layout(buf)
        buf_gen_contents: tuple[BufGenContent, ...] = await fetch_buf_gen_contents(buf_targets, buf)
    else:
        buf_layout = BufLayout("", (), ())
        buf_gen_contents = ()

    plugin_suffixes = {
        **DEFAULT_PLUGIN_SUFFIXES,
        **python_protobuf_subsystem.extra_buf_plugin_suffixes,
    }
    plans: list[_BufStripPlan] = []
    for gen in buf_gen_contents:
        _emit_per_target_warnings_for_buf(gen.target)
        if gen.template_path is None:
            logger.debug(
                "No `buf.gen.yaml` resolved for %s; falling back to source-root path "
                "arithmetic for `_pb2` only. Service suffixes can't be inferred without "
                "the template.",
                gen.target.address,
            )
            plans.append(_fallback_plan(gen.target))
            continue
        # Inference doesn't enforce pinning — that's codegen's job. Inference
        # works fine on unpinned entries (we only need plugin ids to look up
        # suffixes), so the user's editor-side dep inference doesn't fail on
        # in-flight `buf.gen.yaml` edits.
        suffix_outs = parse_plugin_outs(gen.content, plugin_suffixes)
        proto_path = gen.target[ProtobufSourceField].file_path
        buf_module_root = buf_layout.root_for_proto(proto_path)
        plans.append(_plan_buf_target(gen.target, suffix_outs, buf_module_root))

    flat_strip_requests: list[StrippedFileNameRequest] = [
        StrippedFileNameRequest(p) for plan in plans for p in plan.paths_to_strip
    ]
    flat_stripped = (
        await concurrently(strip_file_name(req) for req in flat_strip_requests)
        if flat_strip_requests
        else ()
    )

    # Reassemble per-target module lists.
    buf_modules_per_target: list[list[str]] = []
    idx = 0
    for plan in plans:
        target_modules: list[str] = []
        for suffix in plan.suffixes:
            stripped = flat_stripped[idx]
            target_modules.append(proto_path_to_py_module(stripped.value, suffix=suffix))
            idx += 1
        buf_modules_per_target.append(target_modules)

    # ---- Build the module → providers map. ----
    resolves_to_modules_to_providers: DefaultDict[
        ResolveName, DefaultDict[str, list[ModuleProvider]]
    ] = defaultdict(lambda: defaultdict(list))

    for tgt, stripped_file in zip(protoc_targets, stripped_file_per_protoc_target):
        resolve = tgt[PythonResolveField].normalized_value(python_setup)

        # NB: We don't consider the MyPy plugin, which generates `_pb2.pyi`. The stubs end up
        # sharing the same module as the implementation `_pb2.py`. Because both generated files
        # come from the same original Protobuf target, we're covered.
        module = proto_path_to_py_module(stripped_file.value, suffix="_pb2")
        resolves_to_modules_to_providers[resolve][module].append(
            ModuleProvider(tgt.address, ModuleProviderType.IMPL)
        )
        if tgt.get(ProtobufGrpcToggleField).value:
            for suffix in grpc_suffixes:
                module = proto_path_to_py_module(stripped_file.value, suffix=suffix)
                resolves_to_modules_to_providers[resolve][module].append(
                    ModuleProvider(tgt.address, ModuleProviderType.IMPL)
                )

    for tgt, modules in zip(buf_targets, buf_modules_per_target):
        resolve = tgt[PythonResolveField].normalized_value(python_setup)
        for module in modules:
            resolves_to_modules_to_providers[resolve][module].append(
                ModuleProvider(tgt.address, ModuleProviderType.IMPL)
            )

    # Register BSR-dep Python modules as owned by buf targets that actually
    # generate them. A target only generates BSR-dep `*_pb2.py` files if its
    # `buf.gen.yaml` has `include_imports: true` on whichever plugin emits
    # `_pb2` (the BSR remote, `protoc_builtin: python`, etc.) — otherwise the
    # file isn't in the `GeneratedSources` digest and registering ownership
    # would lie. Targets without `include_imports` are skipped; users either set
    # it or accept the dep-inference warning.
    if buf_layout.deps and buf_targets:
        bsr_to_modules: Mapping[str, Sequence[str]] = {
            **{k: tuple(v) for k, v in DEFAULT_BSR_DEP_MODULES.items()},
            **{k: tuple(v) for k, v in python_protobuf_subsystem.extra_buf_bsr_modules.items()},
        }
        bsr_modules_for_layout: list[str] = []
        for dep in buf_layout.deps:
            bsr_modules_for_layout.extend(bsr_to_modules.get(dep, ()))
        for tgt, gen in zip(buf_targets, buf_gen_contents):
            if not suffix_plugin_includes_imports(gen.content, "_pb2", plugin_suffixes):
                continue
            resolve = tgt[PythonResolveField].normalized_value(python_setup)
            for module in bsr_modules_for_layout:
                resolves_to_modules_to_providers[resolve][module].append(
                    ModuleProvider(tgt.address, ModuleProviderType.IMPL)
                )

    return FirstPartyPythonMappingImpl.create(resolves_to_modules_to_providers)


def rules():
    return (
        *collect_rules(),
        UnionRule(FirstPartyPythonMappingImplMarker, PythonProtobufMappingMarker),
    )

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	4✔
5
6	import logging	4✔
7	import os	4✔
8	from collections import defaultdict	4✔
9	from collections.abc import Mapping, Sequence	4✔
10	from dataclasses import dataclass	4✔
11	from typing import DefaultDict	4✔
12
13	from pants.backend.codegen.protobuf.buf.config import (	4✔
14	BufGenContent,
15	BufLayout,
16	fetch_buf_gen_contents,
17	fetch_buf_layout,
18	parse_plugin_outs,
19	suffix_plugin_includes_imports,
20	)
21	from pants.backend.codegen.protobuf.buf.subsystem import BufSubsystem	4✔
22	from pants.backend.codegen.protobuf.python.additional_fields import PythonSourceRootField	4✔
23	from pants.backend.codegen.protobuf.python.python_protobuf_subsystem import (	4✔
24	DEFAULT_BSR_DEP_MODULES,
25	DEFAULT_PLUGIN_SUFFIXES,
26	PythonProtobufSubsystem,
27	)
28	from pants.backend.codegen.protobuf.target_types import (	4✔
29	AllProtobufTargets,
30	ProtobufGeneratorField,
31	ProtobufGrpcToggleField,
32	ProtobufSourceField,
33	)
34	from pants.backend.python.dependency_inference.module_mapper import (	4✔
35	FirstPartyPythonMappingImpl,
36	FirstPartyPythonMappingImplMarker,
37	ModuleProvider,
38	ModuleProviderType,
39	ResolveName,
40	)
41	from pants.backend.python.subsystems.setup import PythonSetup	4✔
42	from pants.backend.python.target_types import PythonResolveField	4✔
43	from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name	4✔
44	from pants.engine.rules import collect_rules, concurrently, rule	4✔
45	from pants.engine.target import Target	4✔
46	from pants.engine.unions import UnionRule	4✔
47	from pants.util.logging import LogLevel	4✔
48
49	logger = logging.getLogger(__name__)	4✔
50
51
52	def proto_path_to_py_module(stripped_path: str, *, suffix: str) -> str:	4✔
53	return stripped_path.replace(".proto", suffix).replace("/", ".")	2✔
54
55
56	# This is only used to register our implementation with the plugin hook via unions.
57	class PythonProtobufMappingMarker(FirstPartyPythonMappingImplMarker):	4✔
58	pass	4✔
59
60
61	# Suffixes relevant to Python codegen. Each is registered iff its plugin appears
62	# in `buf.gen.yaml`.
63	_PB2_SUFFIX = "_pb2"	4✔
64	_SERVICE_SUFFIXES: tuple[str, ...] = ("_pb2_grpc", "_grpc", "_connect")	4✔
65
66
67	@dataclass(frozen=True)	4✔
68	class _BufStripPlan:	4✔
69	"""Plan for one buf target: paths to feed to `strip_file_name`, paired with the
70	suffix to apply to each stripped result."""
71
72	suffixes: tuple[str, ...]	4✔
73	paths_to_strip: tuple[str, ...]	4✔
74
75
76	def _plan_buf_target(	4✔
77	target: Target,
78	suffix_outs: Mapping[str, str],
79	buf_module_root: str,
80	) -> _BufStripPlan:
81	"""Build the strip plan from the suffixes matched in this target's `buf.gen.yaml`.
82
83	A suffix is registered iff its plugin appears in the file. `grpc=True` is
84	not consulted: `buf.gen.yaml` is the authoritative source of truth for
85	which buf target outputs exist.
86	"""
87	proto_path = target[ProtobufSourceField].file_path	1✔
88	rel_proto = (	1✔
89	os.path.relpath(proto_path, buf_module_root)
90	if buf_module_root
91	and (proto_path == buf_module_root or proto_path.startswith(buf_module_root + os.sep))
92	else proto_path
93	)
94
95	def _path_for(out_dir: str) -> str:	1✔
96	return os.path.normpath(os.path.join(out_dir, rel_proto))	1✔
97
98	suffixes: list[str] = []	1✔
99	paths: list[str] = []	1✔
100	if _PB2_SUFFIX in suffix_outs:	1✔
101	suffixes.append(_PB2_SUFFIX)	1✔
102	paths.append(_path_for(suffix_outs[_PB2_SUFFIX]))	1✔
103	for suffix in _SERVICE_SUFFIXES:	1✔
104	if suffix in suffix_outs:	1✔
105	suffixes.append(suffix)	1✔
106	paths.append(_path_for(suffix_outs[suffix]))	1✔
107
108	return _BufStripPlan(tuple(suffixes), tuple(paths))	1✔
109
110
111	def _fallback_plan(target: Target) -> _BufStripPlan:	4✔
112	"""Plan when no `buf.gen.yaml` was found: assume the proto's source root also
113	covers the generated `.py`. Service suffixes can't be inferred without the
114	template, so we register only `_pb2`."""
115	proto_path = target[ProtobufSourceField].file_path	1✔
116	return _BufStripPlan((_PB2_SUFFIX,), (proto_path,))	1✔
117
118
119	# Protoc-only subsystem options. Their default values are mirrored here so we can
120	# detect when the user explicitly set them while also using buf targets, and warn
121	# that they're ignored on the buf path. Keep in sync with the option definitions
122	# in `python_protobuf_subsystem.py`.
123	_PROTOC_ONLY_OPTION_DEFAULTS: tuple[tuple[str, object], ...] = (	4✔
124	("grpcio_plugin", True),
125	("grpclib_plugin", False),
126	("mypy_plugin", False),
127	("generate_type_stubs", False),
128	)
129
130
131	def _emit_subsystem_warnings_for_buf(subsystem: PythonProtobufSubsystem) -> None:	4✔
132	"""Warn once if subsystem options that are protoc-only are set non-default
133	while at least one buf target exists."""
134	for option_name, default in _PROTOC_ONLY_OPTION_DEFAULTS:	1✔
135	if getattr(subsystem, option_name) == default:	1✔
136	continue	1✔
NEW 137	logger.warning(	×
138	"[%s].%s is set but ignored for `protobuf_generator='buf'` targets. "
139	"Service generation and `.pyi` stubs for buf targets are determined by "
140	"the plugin entries in `buf.gen.yaml`.",
141	subsystem.options_scope,
142	option_name,
143	)
144
145
146	def _emit_per_target_warnings_for_buf(target: Target) -> None:	4✔
147	"""Warn once per buf target about field values that have no effect."""
148	if target.get(ProtobufGrpcToggleField).value:	1✔
149	logger.warning(	1✔
150	"`grpc=True` is set on %s but is ignored for `protobuf_generator='buf'` "
151	"targets. Whether `_pb2_grpc.py` / `_grpc.py` / `_connect.py` exist is "
152	"determined by the plugins in `buf.gen.yaml`.",
153	target.address,
154	)
155	if target.get(PythonSourceRootField).value is not None:	1✔
NEW 156	logger.warning(	×
157	"`python_source_root` is set on %s but ignored for "
158	"`protobuf_generator='buf'`; output paths come from the `out:` field of "
159	"`buf.gen.yaml`.",
160	target.address,
161	)
162
163
164	@rule(desc="Creating map of Protobuf targets to generated Python modules", level=LogLevel.DEBUG)	4✔
165	async def map_protobuf_to_python_modules(	4✔
166	protobuf_targets: AllProtobufTargets,
167	python_setup: PythonSetup,
168	python_protobuf_subsystem: PythonProtobufSubsystem,
169	buf: BufSubsystem,
170	_: PythonProtobufMappingMarker,
171	) -> FirstPartyPythonMappingImpl:
172	grpc_suffixes_list: list[str] = []	2✔
173	if python_protobuf_subsystem.grpcio_plugin:	2✔
174	grpc_suffixes_list.append("_pb2_grpc")	2✔
175	if python_protobuf_subsystem.grpclib_plugin:	2✔
176	grpc_suffixes_list.append("_grpc")	1✔
177	grpc_suffixes = tuple(grpc_suffixes_list)	2✔
178
179	protoc_targets: list[Target] = []	2✔
180	buf_targets: list[Target] = []	2✔
181	for tgt in protobuf_targets:	2✔
182	if tgt.get(ProtobufGeneratorField).value == "buf":	2✔
183	buf_targets.append(tgt)	1✔
184	else:
185	protoc_targets.append(tgt)	2✔
186
187	if buf_targets:	2✔
188	_emit_subsystem_warnings_for_buf(python_protobuf_subsystem)	1✔
189
190	# ---- protoc path. ----
191	stripped_file_per_protoc_target = await concurrently(	2✔
192	strip_file_name(StrippedFileNameRequest(tgt[ProtobufSourceField].file_path))
193	for tgt in protoc_targets
194	)
195
196	# ---- buf path: registry-driven plugin matching, no `grpc=True` gate. ----
197	if buf_targets:	2✔
198	buf_layout: BufLayout = await fetch_buf_layout(buf)	1✔
199	buf_gen_contents: tuple[BufGenContent, ...] = await fetch_buf_gen_contents(buf_targets, buf)	1✔
200	else:
201	buf_layout = BufLayout("", (), ())	2✔
202	buf_gen_contents = ()	2✔
203
204	plugin_suffixes = {	2✔
205	**DEFAULT_PLUGIN_SUFFIXES,
206	**python_protobuf_subsystem.extra_buf_plugin_suffixes,
207	}
208	plans: list[_BufStripPlan] = []	2✔
209	for gen in buf_gen_contents:	2✔
210	_emit_per_target_warnings_for_buf(gen.target)	1✔
211	if gen.template_path is None:	1✔
212	logger.debug(	1✔
213	"No `buf.gen.yaml` resolved for %s; falling back to source-root path "
214	"arithmetic for `_pb2` only. Service suffixes can't be inferred without "
215	"the template.",
216	gen.target.address,
217	)
218	plans.append(_fallback_plan(gen.target))	1✔
219	continue	1✔
220	# Inference doesn't enforce pinning — that's codegen's job. Inference
221	# works fine on unpinned entries (we only need plugin ids to look up
222	# suffixes), so the user's editor-side dep inference doesn't fail on
223	# in-flight `buf.gen.yaml` edits.
224	suffix_outs = parse_plugin_outs(gen.content, plugin_suffixes)	1✔
225	proto_path = gen.target[ProtobufSourceField].file_path	1✔
226	buf_module_root = buf_layout.root_for_proto(proto_path)	1✔
227	plans.append(_plan_buf_target(gen.target, suffix_outs, buf_module_root))	1✔
228
229	flat_strip_requests: list[StrippedFileNameRequest] = [	2✔
230	StrippedFileNameRequest(p) for plan in plans for p in plan.paths_to_strip
231	]
232	flat_stripped = (	2✔
233	await concurrently(strip_file_name(req) for req in flat_strip_requests)
234	if flat_strip_requests
235	else ()
236	)
237
238	# Reassemble per-target module lists.
239	buf_modules_per_target: list[list[str]] = []	2✔
240	idx = 0	2✔
241	for plan in plans:	2✔
242	target_modules: list[str] = []	1✔
243	for suffix in plan.suffixes:	1✔
244	stripped = flat_stripped[idx]	1✔
245	target_modules.append(proto_path_to_py_module(stripped.value, suffix=suffix))	1✔
246	idx += 1	1✔
247	buf_modules_per_target.append(target_modules)	1✔
248
249	# ---- Build the module → providers map. ----
250	resolves_to_modules_to_providers: DefaultDict[	2✔
251	ResolveName, DefaultDict[str, list[ModuleProvider]]
252	] = defaultdict(lambda: defaultdict(list))
253
254	for tgt, stripped_file in zip(protoc_targets, stripped_file_per_protoc_target):	2✔
255	resolve = tgt[PythonResolveField].normalized_value(python_setup)	2✔
256
257	# NB: We don't consider the MyPy plugin, which generates `_pb2.pyi`. The stubs end up
258	# sharing the same module as the implementation `_pb2.py`. Because both generated files
259	# come from the same original Protobuf target, we're covered.
260	module = proto_path_to_py_module(stripped_file.value, suffix="_pb2")	2✔
261	resolves_to_modules_to_providers[resolve][module].append(	2✔
262	ModuleProvider(tgt.address, ModuleProviderType.IMPL)
263	)
264	if tgt.get(ProtobufGrpcToggleField).value:	2✔
265	for suffix in grpc_suffixes:	1✔
266	module = proto_path_to_py_module(stripped_file.value, suffix=suffix)	1✔
267	resolves_to_modules_to_providers[resolve][module].append(	1✔
268	ModuleProvider(tgt.address, ModuleProviderType.IMPL)
269	)
270
271	for tgt, modules in zip(buf_targets, buf_modules_per_target):	2✔
272	resolve = tgt[PythonResolveField].normalized_value(python_setup)	1✔
273	for module in modules:	1✔
274	resolves_to_modules_to_providers[resolve][module].append(	1✔
275	ModuleProvider(tgt.address, ModuleProviderType.IMPL)
276	)
277
278	# Register BSR-dep Python modules as owned by buf targets that actually
279	# generate them. A target only generates BSR-dep `*_pb2.py` files if its
280	# `buf.gen.yaml` has `include_imports: true` on whichever plugin emits
281	# `_pb2` (the BSR remote, `protoc_builtin: python`, etc.) — otherwise the
282	# file isn't in the `GeneratedSources` digest and registering ownership
283	# would lie. Targets without `include_imports` are skipped; users either set
284	# it or accept the dep-inference warning.
285	if buf_layout.deps and buf_targets:	2✔
286	bsr_to_modules: Mapping[str, Sequence[str]] = {	1✔
287	**{k: tuple(v) for k, v in DEFAULT_BSR_DEP_MODULES.items()},
288	**{k: tuple(v) for k, v in python_protobuf_subsystem.extra_buf_bsr_modules.items()},
289	}
290	bsr_modules_for_layout: list[str] = []	1✔
291	for dep in buf_layout.deps:	1✔
292	bsr_modules_for_layout.extend(bsr_to_modules.get(dep, ()))	1✔
293	for tgt, gen in zip(buf_targets, buf_gen_contents):	1✔
294	if not suffix_plugin_includes_imports(gen.content, "_pb2", plugin_suffixes):	1✔
295	continue	1✔
296	resolve = tgt[PythonResolveField].normalized_value(python_setup)	1✔
297	for module in bsr_modules_for_layout:	1✔
298	resolves_to_modules_to_providers[resolve][module].append(	1✔
299	ModuleProvider(tgt.address, ModuleProviderType.IMPL)
300	)
301
302	return FirstPartyPythonMappingImpl.create(resolves_to_modules_to_providers)	2✔
303
304
305	def rules():	4✔
306	return (	4✔
307	*collect_rules(),
308	UnionRule(FirstPartyPythonMappingImplMarker, PythonProtobufMappingMarker),
309	)

pantsbuild / pants / 26689585807

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