23068140808

Committed 13 Mar 2026 07:56PM UTC coverage: 52.677% (-40.3%) from 92.932%

Build # 23068140808

Build Type

Pull #23170

github

Committed by

web-flow

Commit Message

Merge e8ca01cfa into f07276df6

Pull Request Pull Request #23170: Debug reapi test cache misses

Run Details

31687 of 60153 relevant lines covered (52.68%)

1.05 hits per line

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

97.37

/src/python/pants/backend/python/typecheck/pytype/rules.py

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

from __future__ import annotations

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

from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.target_types import (
    InterpreterConstraintsField,
    PythonResolveField,
    PythonSourceField,
)
from pants.backend.python.typecheck.pytype.skip_field import SkipPytypeField
from pants.backend.python.typecheck.pytype.subsystem import Pytype
from pants.backend.python.util_rules import pex_from_targets
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.partition import (
    _partition_by_interpreter_constraints_and_resolve,
)
from pants.backend.python.util_rules.pex import (
    PexRequest,
    VenvPexProcess,
    VenvPexRequest,
    create_pex,
    create_venv_pex,
)
from pants.backend.python.util_rules.pex_environment import PexEnvironment
from pants.backend.python.util_rules.pex_from_targets import RequirementsPexRequest
from pants.core.goals.check import CheckRequest, CheckResult, CheckResults, CheckSubsystem
from pants.core.goals.resolves import ExportableTool
from pants.core.util_rules import config_files
from pants.core.util_rules.config_files import find_config_file
from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files
from pants.engine.collection import Collection
from pants.engine.internals.graph import resolve_coarsened_targets as coarsened_targets_get
from pants.engine.internals.native_engine import MergeDigests
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import execute_process, merge_digests
from pants.engine.rules import Rule, collect_rules, implicitly, rule
from pants.engine.target import CoarsenedTargets, CoarsenedTargetsRequest, FieldSet, Target
from pants.engine.unions import UnionRule
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet, OrderedSet
from pants.util.strutil import pluralize

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class PytypeFieldSet(FieldSet):
    required_fields = (PythonSourceField,)

    sources: PythonSourceField
    resolve: PythonResolveField
    interpreter_constraints: InterpreterConstraintsField

    @classmethod
    def opt_out(cls, tgt: Target) -> bool:
        return tgt.get(SkipPytypeField).value


class PytypeRequest(CheckRequest):
    field_set_type = PytypeFieldSet
    tool_name = Pytype.options_scope


@dataclass(frozen=True)
class PytypePartition:
    field_sets: FrozenOrderedSet[PytypeFieldSet]
    root_targets: CoarsenedTargets
    resolve_description: str | None
    interpreter_constraints: InterpreterConstraints

    def description(self) -> str:
        ics = str(sorted(str(c) for c in self.interpreter_constraints))
        return f"{self.resolve_description}, {ics}" if self.resolve_description else ics


class PytypePartitions(Collection[PytypePartition]):
    pass


@rule(
    desc="Pytype typecheck each partition based on its interpreter_constraints",
    level=LogLevel.DEBUG,
)
async def pytype_typecheck_partition(
    partition: PytypePartition,
    pytype: Pytype,
    check_subsystem: CheckSubsystem,
    pex_environment: PexEnvironment,
) -> CheckResult:
    roots_sources, requirements_pex, pytype_pex, config_files = await concurrently(
        determine_source_files(SourceFilesRequest(fs.sources for fs in partition.field_sets)),
        create_pex(
            **implicitly(
                RequirementsPexRequest(
                    (fs.address for fs in partition.field_sets),
                    hardcoded_interpreter_constraints=partition.interpreter_constraints,
                )
            )
        ),
        create_pex(
            pytype.to_pex_request(interpreter_constraints=partition.interpreter_constraints)
        ),
        find_config_file(pytype.config_request()),
    )

    input_digest = await merge_digests(
        MergeDigests((roots_sources.snapshot.digest, config_files.snapshot.digest))
    )

    runner = await create_venv_pex(
        VenvPexRequest(
            PexRequest(
                output_filename="pytype_runner.pex",
                interpreter_constraints=partition.interpreter_constraints,
                main=pytype.main,
                internal_only=True,
                pex_path=[pytype_pex, requirements_pex],
            ),
            pex_environment.in_sandbox(working_directory=None),
        ),
        **implicitly(),
    )

    result = await execute_process(
        **implicitly(
            VenvPexProcess(
                runner,
                argv=(
                    *(("--config", pytype.config) if pytype.config else ()),
                    "{pants_concurrency}",
                    *pytype.args,
                    *roots_sources.files,
                ),
                # This adds the venv/bin folder to PATH
                extra_env={
                    "PEX_VENV_BIN_PATH": "prepend",
                },
                input_digest=input_digest,
                output_files=roots_sources.files,
                concurrency_available=len(roots_sources.files),
                description=f"Run Pytype on {pluralize(len(roots_sources.files), 'file')}.",
                level=LogLevel.DEBUG,
                cache_scope=check_subsystem.default_process_cache_scope,
            )
        )
    )

    return CheckResult.from_fallible_process_result(
        result,
        partition_description=partition.description(),
    )


@rule(
    desc="Determine if it is necessary to partition Pytype's input (interpreter_constraints and resolves)",
    level=LogLevel.DEBUG,
)
async def pytype_determine_partitions(
    request: PytypeRequest,
    pytype: Pytype,
    python_setup: PythonSetup,
) -> PytypePartitions:
    resolve_and_interpreter_constraints_to_field_sets = (
        _partition_by_interpreter_constraints_and_resolve(request.field_sets, python_setup)
    )

    coarsened_targets = await coarsened_targets_get(
        CoarsenedTargetsRequest(field_set.address for field_set in request.field_sets),
        **implicitly(),
    )
    coarsened_targets_by_address = coarsened_targets.by_address()

    return PytypePartitions(
        PytypePartition(
            FrozenOrderedSet(field_sets),
            CoarsenedTargets(
                OrderedSet(
                    coarsened_targets_by_address[field_set.address] for field_set in field_sets
                )
            ),
            resolve if len(python_setup.resolves) > 1 else None,
            interpreter_constraints or pytype.interpreter_constraints,
        )
        for (resolve, interpreter_constraints), field_sets in sorted(
            resolve_and_interpreter_constraints_to_field_sets.items()
        )
    )


@rule(desc="Typecheck using Pytype", level=LogLevel.DEBUG)
async def pytype_typecheck(
    request: PytypeRequest,
    pytype: Pytype,
) -> CheckResults:
    if pytype.skip:
        return CheckResults([], checker_name=request.tool_name)

    # Explicitly excluding `pytype` as a function argument to `pytype_determine_partitions` and `pytype_typecheck_partition`
    # as it throws "TypeError: unhashable type: 'Pytype'"
    partitions = await pytype_determine_partitions(request, **implicitly())
    partitioned_results = await concurrently(
        pytype_typecheck_partition(partition, **implicitly()) for partition in partitions
    )
    return CheckResults(
        partitioned_results,
        checker_name=request.tool_name,
    )


def rules() -> Iterable[Rule | UnionRule]:
    return (
        *collect_rules(),
        *config_files.rules(),
        *pex_from_targets.rules(),
        UnionRule(CheckRequest, PytypeRequest),
        UnionRule(ExportableTool, Pytype),
    )

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	2✔
5
6	import logging	2✔
7	from collections.abc import Iterable	2✔
8	from dataclasses import dataclass	2✔
9
10	from pants.backend.python.subsystems.setup import PythonSetup	2✔
11	from pants.backend.python.target_types import (	2✔
12	InterpreterConstraintsField,
13	PythonResolveField,
14	PythonSourceField,
15	)
16	from pants.backend.python.typecheck.pytype.skip_field import SkipPytypeField	2✔
17	from pants.backend.python.typecheck.pytype.subsystem import Pytype	2✔
18	from pants.backend.python.util_rules import pex_from_targets	2✔
19	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	2✔
20	from pants.backend.python.util_rules.partition import (	2✔
21	_partition_by_interpreter_constraints_and_resolve,
22	)
23	from pants.backend.python.util_rules.pex import (	2✔
24	PexRequest,
25	VenvPexProcess,
26	VenvPexRequest,
27	create_pex,
28	create_venv_pex,
29	)
30	from pants.backend.python.util_rules.pex_environment import PexEnvironment	2✔
31	from pants.backend.python.util_rules.pex_from_targets import RequirementsPexRequest	2✔
32	from pants.core.goals.check import CheckRequest, CheckResult, CheckResults, CheckSubsystem	2✔
33	from pants.core.goals.resolves import ExportableTool	2✔
34	from pants.core.util_rules import config_files	2✔
35	from pants.core.util_rules.config_files import find_config_file	2✔
36	from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files	2✔
37	from pants.engine.collection import Collection	2✔
38	from pants.engine.internals.graph import resolve_coarsened_targets as coarsened_targets_get	2✔
39	from pants.engine.internals.native_engine import MergeDigests	2✔
40	from pants.engine.internals.selectors import concurrently	2✔
41	from pants.engine.intrinsics import execute_process, merge_digests	2✔
42	from pants.engine.rules import Rule, collect_rules, implicitly, rule	2✔
43	from pants.engine.target import CoarsenedTargets, CoarsenedTargetsRequest, FieldSet, Target	2✔
44	from pants.engine.unions import UnionRule	2✔
45	from pants.util.logging import LogLevel	2✔
46	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	2✔
47	from pants.util.strutil import pluralize	2✔
48
49	logger = logging.getLogger(__name__)	2✔
50
51
52	@dataclass(frozen=True)	2✔
53	class PytypeFieldSet(FieldSet):	2✔
54	required_fields = (PythonSourceField,)	2✔
55
56	sources: PythonSourceField	2✔
57	resolve: PythonResolveField	2✔
58	interpreter_constraints: InterpreterConstraintsField	2✔
59
60	@classmethod	2✔
61	def opt_out(cls, tgt: Target) -> bool:	2✔
62	return tgt.get(SkipPytypeField).value	×
63
64
65	class PytypeRequest(CheckRequest):	2✔
66	field_set_type = PytypeFieldSet	2✔
67	tool_name = Pytype.options_scope	2✔
68
69
70	@dataclass(frozen=True)	2✔
71	class PytypePartition:	2✔
72	field_sets: FrozenOrderedSet[PytypeFieldSet]	2✔
73	root_targets: CoarsenedTargets	2✔
74	resolve_description: str \| None	2✔
75	interpreter_constraints: InterpreterConstraints	2✔
76
77	def description(self) -> str:	2✔
78	ics = str(sorted(str(c) for c in self.interpreter_constraints))	2✔
79	return f"{self.resolve_description}, {ics}" if self.resolve_description else ics	2✔
80
81
82	class PytypePartitions(Collection[PytypePartition]):	2✔
83	pass	2✔
84
85
86	@rule(	2✔
87	desc="Pytype typecheck each partition based on its interpreter_constraints",
88	level=LogLevel.DEBUG,
89	)
90	async def pytype_typecheck_partition(	2✔
91	partition: PytypePartition,
92	pytype: Pytype,
93	check_subsystem: CheckSubsystem,
94	pex_environment: PexEnvironment,
95	) -> CheckResult:
96	roots_sources, requirements_pex, pytype_pex, config_files = await concurrently(	2✔
97	determine_source_files(SourceFilesRequest(fs.sources for fs in partition.field_sets)),
98	create_pex(
99	**implicitly(
100	RequirementsPexRequest(
101	(fs.address for fs in partition.field_sets),
102	hardcoded_interpreter_constraints=partition.interpreter_constraints,
103	)
104	)
105	),
106	create_pex(
107	pytype.to_pex_request(interpreter_constraints=partition.interpreter_constraints)
108	),
109	find_config_file(pytype.config_request()),
110	)
111
112	input_digest = await merge_digests(	2✔
113	MergeDigests((roots_sources.snapshot.digest, config_files.snapshot.digest))
114	)
115
116	runner = await create_venv_pex(	2✔
117	VenvPexRequest(
118	PexRequest(
119	output_filename="pytype_runner.pex",
120	interpreter_constraints=partition.interpreter_constraints,
121	main=pytype.main,
122	internal_only=True,
123	pex_path=[pytype_pex, requirements_pex],
124	),
125	pex_environment.in_sandbox(working_directory=None),
126	),
127	**implicitly(),
128	)
129
130	result = await execute_process(	2✔
131	**implicitly(
132	VenvPexProcess(
133	runner,
134	argv=(
135	*(("--config", pytype.config) if pytype.config else ()),
136	"{pants_concurrency}",
137	*pytype.args,
138	*roots_sources.files,
139	),
140	# This adds the venv/bin folder to PATH
141	extra_env={
142	"PEX_VENV_BIN_PATH": "prepend",
143	},
144	input_digest=input_digest,
145	output_files=roots_sources.files,
146	concurrency_available=len(roots_sources.files),
147	description=f"Run Pytype on {pluralize(len(roots_sources.files), 'file')}.",
148	level=LogLevel.DEBUG,
149	cache_scope=check_subsystem.default_process_cache_scope,
150	)
151	)
152	)
153
154	return CheckResult.from_fallible_process_result(	2✔
155	result,
156	partition_description=partition.description(),
157	)
158
159
160	@rule(	2✔
161	desc="Determine if it is necessary to partition Pytype's input (interpreter_constraints and resolves)",
162	level=LogLevel.DEBUG,
163	)
164	async def pytype_determine_partitions(	2✔
165	request: PytypeRequest,
166	pytype: Pytype,
167	python_setup: PythonSetup,
168	) -> PytypePartitions:
169	resolve_and_interpreter_constraints_to_field_sets = (	2✔
170	_partition_by_interpreter_constraints_and_resolve(request.field_sets, python_setup)
171	)
172
173	coarsened_targets = await coarsened_targets_get(	2✔
174	CoarsenedTargetsRequest(field_set.address for field_set in request.field_sets),
175	**implicitly(),
176	)
177	coarsened_targets_by_address = coarsened_targets.by_address()	2✔
178
179	return PytypePartitions(	2✔
180	PytypePartition(
181	FrozenOrderedSet(field_sets),
182	CoarsenedTargets(
183	OrderedSet(
184	coarsened_targets_by_address[field_set.address] for field_set in field_sets
185	)
186	),
187	resolve if len(python_setup.resolves) > 1 else None,
188	interpreter_constraints or pytype.interpreter_constraints,
189	)
190	for (resolve, interpreter_constraints), field_sets in sorted(
191	resolve_and_interpreter_constraints_to_field_sets.items()
192	)
193	)
194
195
196	@rule(desc="Typecheck using Pytype", level=LogLevel.DEBUG)	2✔
197	async def pytype_typecheck(	2✔
198	request: PytypeRequest,
199	pytype: Pytype,
200	) -> CheckResults:
201	if pytype.skip:	2✔
202	return CheckResults([], checker_name=request.tool_name)	×
203
204	# Explicitly excluding `pytype` as a function argument to `pytype_determine_partitions` and `pytype_typecheck_partition`
205	# as it throws "TypeError: unhashable type: 'Pytype'"
206	partitions = await pytype_determine_partitions(request, **implicitly())	2✔
207	partitioned_results = await concurrently(	2✔
208	pytype_typecheck_partition(partition, **implicitly()) for partition in partitions
209	)
210	return CheckResults(	2✔
211	partitioned_results,
212	checker_name=request.tool_name,
213	)
214
215
216	def rules() -> Iterable[Rule \| UnionRule]:	2✔
217	return (	2✔
218	*collect_rules(),
219	*config_files.rules(),
220	*pex_from_targets.rules(),
221	UnionRule(CheckRequest, PytypeRequest),
222	UnionRule(ExportableTool, Pytype),
223	)

pantsbuild / pants / 23068140808

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