22523112068

Committed 28 Feb 2026 03:01PM UTC coverage: 90.325% (-2.6%) from 92.93%

Build # 22523112068

Build Type

push

github

Committed by

web-flow

Commit Message

Prepare 2.32.0.dev3 (#23148)

Run Details

82731 of 91593 relevant lines covered (90.32%)

3.28 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	1✔
5
6	import logging	1✔
7	from collections.abc import Iterable	1✔
8	from dataclasses import dataclass	1✔
9
10	from pants.backend.python.subsystems.setup import PythonSetup	1✔
11	from pants.backend.python.target_types import (	1✔
12	InterpreterConstraintsField,
13	PythonResolveField,
14	PythonSourceField,
15	)
16	from pants.backend.python.typecheck.pytype.skip_field import SkipPytypeField	1✔
17	from pants.backend.python.typecheck.pytype.subsystem import Pytype	1✔
18	from pants.backend.python.util_rules import pex_from_targets	1✔
19	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	1✔
20	from pants.backend.python.util_rules.partition import (	1✔
21	_partition_by_interpreter_constraints_and_resolve,
22	)
23	from pants.backend.python.util_rules.pex import (	1✔
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	1✔
31	from pants.backend.python.util_rules.pex_from_targets import RequirementsPexRequest	1✔
32	from pants.core.goals.check import CheckRequest, CheckResult, CheckResults, CheckSubsystem	1✔
33	from pants.core.goals.resolves import ExportableTool	1✔
34	from pants.core.util_rules import config_files	1✔
35	from pants.core.util_rules.config_files import find_config_file	1✔
36	from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files	1✔
37	from pants.engine.collection import Collection	1✔
38	from pants.engine.internals.graph import resolve_coarsened_targets as coarsened_targets_get	1✔
39	from pants.engine.internals.native_engine import MergeDigests	1✔
40	from pants.engine.internals.selectors import concurrently	1✔
41	from pants.engine.intrinsics import execute_process, merge_digests	1✔
42	from pants.engine.rules import Rule, collect_rules, implicitly, rule	1✔
43	from pants.engine.target import CoarsenedTargets, CoarsenedTargetsRequest, FieldSet, Target	1✔
44	from pants.engine.unions import UnionRule	1✔
45	from pants.util.logging import LogLevel	1✔
46	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	1✔
47	from pants.util.strutil import pluralize	1✔
48
49	logger = logging.getLogger(__name__)	1✔
50
51
52	@dataclass(frozen=True)	1✔
53	class PytypeFieldSet(FieldSet):	1✔
54	required_fields = (PythonSourceField,)	1✔
55
56	sources: PythonSourceField	1✔
57	resolve: PythonResolveField	1✔
58	interpreter_constraints: InterpreterConstraintsField	1✔
59
60	@classmethod	1✔
61	def opt_out(cls, tgt: Target) -> bool:	1✔
62	return tgt.get(SkipPytypeField).value	×
63
64
65	class PytypeRequest(CheckRequest):	1✔
66	field_set_type = PytypeFieldSet	1✔
67	tool_name = Pytype.options_scope	1✔
68
69
70	@dataclass(frozen=True)	1✔
71	class PytypePartition:	1✔
72	field_sets: FrozenOrderedSet[PytypeFieldSet]	1✔
73	root_targets: CoarsenedTargets	1✔
74	resolve_description: str \| None	1✔
75	interpreter_constraints: InterpreterConstraints	1✔
76
77	def description(self) -> str:	1✔
78	ics = str(sorted(str(c) for c in self.interpreter_constraints))	1✔
79	return f"{self.resolve_description}, {ics}" if self.resolve_description else ics	1✔
80
81
82	class PytypePartitions(Collection[PytypePartition]):	1✔
83	pass	1✔
84
85
86	@rule(	1✔
87	desc="Pytype typecheck each partition based on its interpreter_constraints",
88	level=LogLevel.DEBUG,
89	)
90	async def pytype_typecheck_partition(	1✔
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(	1✔
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(	1✔
113	MergeDigests((roots_sources.snapshot.digest, config_files.snapshot.digest))
114	)
115
116	runner = await create_venv_pex(	1✔
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(	1✔
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(	1✔
155	result,
156	partition_description=partition.description(),
157	)
158
159
160	@rule(	1✔
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(	1✔
165	request: PytypeRequest,
166	pytype: Pytype,
167	python_setup: PythonSetup,
168	) -> PytypePartitions:
169	resolve_and_interpreter_constraints_to_field_sets = (	1✔
170	_partition_by_interpreter_constraints_and_resolve(request.field_sets, python_setup)
171	)
172
173	coarsened_targets = await coarsened_targets_get(	1✔
174	CoarsenedTargetsRequest(field_set.address for field_set in request.field_sets),
175	**implicitly(),
176	)
177	coarsened_targets_by_address = coarsened_targets.by_address()	1✔
178
179	return PytypePartitions(	1✔
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)	1✔
197	async def pytype_typecheck(	1✔
198	request: PytypeRequest,
199	pytype: Pytype,
200	) -> CheckResults:
201	if pytype.skip:	1✔
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())	1✔
207	partitioned_results = await concurrently(	1✔
208	pytype_typecheck_partition(partition, **implicitly()) for partition in partitions
209	)
210	return CheckResults(	1✔
211	partitioned_results,
212	checker_name=request.tool_name,
213	)
214
215
216	def rules() -> Iterable[Rule \| UnionRule]:	1✔
217	return (	1✔
218	*collect_rules(),
219	*config_files.rules(),
220	*pex_from_targets.rules(),
221	UnionRule(CheckRequest, PytypeRequest),
222	UnionRule(ExportableTool, Pytype),
223	)

pantsbuild / pants / 22523112068

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