25259185675

Committed 02 May 2026 06:47PM UTC coverage: 92.141% (-0.8%) from 92.955%

Build # 25259185675

Build Type

push

github

Committed by

web-flow

Commit Message

Fix the dynamic UI. (#23306)

In #23114 we upgraded to indicatif 0.18.4,
which included a fix to respect TERM, and 
display nothing if it's unset.

Since we did not pass TERM through pantsd, the
dynamic ui is now not shown. 

This change fixes that, and also pass NO_COLOR
through, since indicatif inspects it too.

Coverage Stats

88773 of 96345 relevant lines covered (92.14%)

3.83 hits per line

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

97.33

/src/python/pants/backend/python/lint/pylint/rules.py

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

from __future__ import annotations

from dataclasses import dataclass

import packaging

from pants.backend.python.lint.pylint.subsystem import (
    Pylint,
    PylintFieldSet,
    PylintFirstPartyPlugins,
)
from pants.backend.python.subsystems.setup import PythonSetup
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 (
    Pex,
    PexRequest,
    VenvPexProcess,
    VenvPexRequest,
    create_pex,
    create_venv_pex,
    determine_pex_resolve_info,
)
from pants.backend.python.util_rules.pex_environment import PexEnvironment
from pants.backend.python.util_rules.pex_from_targets import RequirementsPexRequest
from pants.backend.python.util_rules.python_sources import (
    PythonSourceFilesRequest,
    prepare_python_sources,
)
from pants.core.goals.lint import REPORT_DIR, LintResult, LintTargetsRequest, Partitions
from pants.core.util_rules.config_files import find_config_file
from pants.core.util_rules.partitions import Partition
from pants.engine.fs import CreateDigest, Directory, MergeDigests, RemovePrefix
from pants.engine.internals.graph import resolve_coarsened_targets as coarsened_targets_get
from pants.engine.intrinsics import create_digest, execute_process, merge_digests, remove_prefix
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import CoarsenedTargets, CoarsenedTargetsRequest
from pants.util.logging import LogLevel
from pants.util.strutil import pluralize


@dataclass(frozen=True)
class PartitionMetadata:
    coarsened_targets: CoarsenedTargets
    # NB: These are the same across every element in a partition
    resolve_description: str | None
    interpreter_constraints: InterpreterConstraints

    @property
    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 PylintRequest(LintTargetsRequest):
    field_set_type = PylintFieldSet
    tool_subsystem = Pylint  # type: ignore[assignment]


def generate_argv(field_sets: tuple[PylintFieldSet, ...], pylint: Pylint) -> tuple[str, ...]:
    args: list[str] = []
    if pylint.config is not None:
        args.append(f"--rcfile={pylint.config}")
    args.append("--jobs={pants_concurrency}")
    args.extend(pylint.args)
    args.extend(field_set.source.file_path for field_set in field_sets)
    return tuple(args)


@rule(desc="Determine if necessary to partition Pylint input", level=LogLevel.DEBUG)
async def partition_pylint(
    request: PylintRequest.PartitionRequest[PylintFieldSet],
    pylint: Pylint,
    python_setup: PythonSetup,
    first_party_plugins: PylintFirstPartyPlugins,
) -> Partitions[PylintFieldSet, PartitionMetadata]:
    if pylint.skip:
        return Partitions()

    first_party_ics = InterpreterConstraints.create_from_compatibility_fields(
        first_party_plugins.interpreter_constraints_and_resolve_fields, python_setup
    )

    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 Partitions(
        Partition(
            tuple(field_sets),
            PartitionMetadata(
                CoarsenedTargets(
                    coarsened_targets_by_address[field_set.address] for field_set in field_sets
                ),
                resolve if len(python_setup.resolves) > 1 else None,
                InterpreterConstraints.merge((interpreter_constraints, first_party_ics)),
            ),
        )
        for (
            resolve,
            interpreter_constraints,
        ), field_sets in resolve_and_interpreter_constraints_to_field_sets.items()
    )


@rule(desc="Lint using Pylint", level=LogLevel.DEBUG)
async def run_pylint(
    request: PylintRequest.Batch[PylintFieldSet, PartitionMetadata],
    pylint: Pylint,
    first_party_plugins: PylintFirstPartyPlugins,
    pex_environment: PexEnvironment,
) -> LintResult:
    assert request.partition_metadata is not None

    # The coarsened targets in the incoming request are for all targets in the request's original
    # partition. Since the core `lint` logic re-batches inputs according to `[lint].batch_size`,
    # this could be many more targets than are actually needed to lint the specific batch of files
    # received by this rule. Subset the CTs one more time here to only those that are relevant.
    all_coarsened_targets_by_address = request.partition_metadata.coarsened_targets.by_address()
    coarsened_targets = CoarsenedTargets(
        all_coarsened_targets_by_address[field_set.address] for field_set in request.elements
    )
    coarsened_closure = tuple(coarsened_targets.closure())

    requirements_pex_get = create_pex(
        **implicitly(
            RequirementsPexRequest(
                (target.address for target in coarsened_closure),
                # NB: These constraints must be identical to the other PEXes. Otherwise, we risk using
                # a different version for the requirements than the other two PEXes, which can result
                # in a PEX runtime error about missing dependencies.
                hardcoded_interpreter_constraints=request.partition_metadata.interpreter_constraints,
            )
        )
    )

    pylint_pex_get = create_pex(
        pylint.to_pex_request(
            interpreter_constraints=request.partition_metadata.interpreter_constraints,
            extra_requirements=first_party_plugins.requirement_strings,
        )
    )

    sources_get = prepare_python_sources(
        PythonSourceFilesRequest(coarsened_closure), **implicitly()
    )
    # Ensure that the empty report dir exists.
    report_directory_digest_get = create_digest(CreateDigest([Directory(REPORT_DIR)]))

    (
        pylint_pex,
        requirements_pex,
        sources,
        report_directory,
    ) = await concurrently(
        pylint_pex_get,
        requirements_pex_get,
        sources_get,
        report_directory_digest_get,
    )

    pylint_pex_info = await determine_pex_resolve_info(**implicitly({pylint_pex: Pex}))
    astroid_info = pylint_pex_info.find("astroid")
    # Astroid is a transitive dependency of pylint and should always be available in the pex.
    assert astroid_info

    pylint_runner_pex, config_files = await concurrently(
        create_venv_pex(
            VenvPexRequest(
                PexRequest(
                    output_filename="pylint_runner.pex",
                    interpreter_constraints=request.partition_metadata.interpreter_constraints,
                    main=pylint.main,
                    internal_only=True,
                    pex_path=[pylint_pex, requirements_pex],
                ),
                pex_environment.in_sandbox(working_directory=None),
                # Astroid < 2.9.1 had a regression that prevented the use of symlinks:
                # https://github.com/PyCQA/pylint/issues/1470
                site_packages_copies=(astroid_info.version < packaging.version.Version("2.9.1")),
            ),
            **implicitly(),
        ),
        find_config_file(pylint.config_request(sources.source_files.snapshot.dirs)),
    )

    pythonpath = list(sources.source_roots)
    if first_party_plugins:
        pythonpath.append(first_party_plugins.PREFIX)

    input_digest = await merge_digests(
        MergeDigests(
            (
                config_files.snapshot.digest,
                first_party_plugins.sources_digest,
                sources.source_files.snapshot.digest,
                report_directory,
            )
        )
    )

    result = await execute_process(
        **implicitly(
            VenvPexProcess(
                pylint_runner_pex,
                argv=generate_argv(request.elements, pylint),
                input_digest=input_digest,
                output_directories=(REPORT_DIR,),
                extra_env={"PEX_EXTRA_SYS_PATH": ":".join(pythonpath)},
                concurrency_available=len(request.elements),
                description=f"Run Pylint on {pluralize(len(request.elements), 'target')}.",
                level=LogLevel.DEBUG,
            )
        )
    )
    report = await remove_prefix(RemovePrefix(result.output_digest, REPORT_DIR))
    return LintResult.create(request, result, report=report)


def rules():
    return (
        *collect_rules(),
        *PylintRequest.rules(),
        *pex_from_targets.rules(),
    )

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	2✔
5
6	from dataclasses import dataclass	2✔
7
8	import packaging	2✔
9
10	from pants.backend.python.lint.pylint.subsystem import (	2✔
11	Pylint,
12	PylintFieldSet,
13	PylintFirstPartyPlugins,
14	)
15	from pants.backend.python.subsystems.setup import PythonSetup	2✔
16	from pants.backend.python.util_rules import pex_from_targets	2✔
17	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	2✔
18	from pants.backend.python.util_rules.partition import (	2✔
19	_partition_by_interpreter_constraints_and_resolve,
20	)
21	from pants.backend.python.util_rules.pex import (	2✔
22	Pex,
23	PexRequest,
24	VenvPexProcess,
25	VenvPexRequest,
26	create_pex,
27	create_venv_pex,
28	determine_pex_resolve_info,
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.backend.python.util_rules.python_sources import (	2✔
33	PythonSourceFilesRequest,
34	prepare_python_sources,
35	)
36	from pants.core.goals.lint import REPORT_DIR, LintResult, LintTargetsRequest, Partitions	2✔
37	from pants.core.util_rules.config_files import find_config_file	2✔
38	from pants.core.util_rules.partitions import Partition	2✔
39	from pants.engine.fs import CreateDigest, Directory, MergeDigests, RemovePrefix	2✔
40	from pants.engine.internals.graph import resolve_coarsened_targets as coarsened_targets_get	2✔
41	from pants.engine.intrinsics import create_digest, execute_process, merge_digests, remove_prefix	2✔
42	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	2✔
43	from pants.engine.target import CoarsenedTargets, CoarsenedTargetsRequest	2✔
44	from pants.util.logging import LogLevel	2✔
45	from pants.util.strutil import pluralize	2✔
46
47
48	@dataclass(frozen=True)	2✔
49	class PartitionMetadata:	2✔
50	coarsened_targets: CoarsenedTargets	2✔
51	# NB: These are the same across every element in a partition
52	resolve_description: str \| None	2✔
53	interpreter_constraints: InterpreterConstraints	2✔
54
55	@property	2✔
56	def description(self) -> str:	2✔
57	ics = str(sorted(str(c) for c in self.interpreter_constraints))	2✔
58	return f"{self.resolve_description}, {ics}" if self.resolve_description else ics	2✔
59
60
61	class PylintRequest(LintTargetsRequest):	2✔
62	field_set_type = PylintFieldSet	2✔
63	tool_subsystem = Pylint # type: ignore[assignment]	2✔
64
65
66	def generate_argv(field_sets: tuple[PylintFieldSet, ...], pylint: Pylint) -> tuple[str, ...]:	2✔
67	args: list[str] = []	2✔
68	if pylint.config is not None:	2✔
69	args.append(f"--rcfile={pylint.config}")	×
70	args.append("--jobs={pants_concurrency}")	2✔
71	args.extend(pylint.args)	2✔
72	args.extend(field_set.source.file_path for field_set in field_sets)	2✔
73	return tuple(args)	2✔
74
75
76	@rule(desc="Determine if necessary to partition Pylint input", level=LogLevel.DEBUG)	2✔
77	async def partition_pylint(	2✔
78	request: PylintRequest.PartitionRequest[PylintFieldSet],
79	pylint: Pylint,
80	python_setup: PythonSetup,
81	first_party_plugins: PylintFirstPartyPlugins,
82	) -> Partitions[PylintFieldSet, PartitionMetadata]:
83	if pylint.skip:	2✔
84	return Partitions()	×
85
86	first_party_ics = InterpreterConstraints.create_from_compatibility_fields(	2✔
87	first_party_plugins.interpreter_constraints_and_resolve_fields, python_setup
88	)
89
90	resolve_and_interpreter_constraints_to_field_sets = (	2✔
91	_partition_by_interpreter_constraints_and_resolve(request.field_sets, python_setup)
92	)
93
94	coarsened_targets = await coarsened_targets_get(	2✔
95	CoarsenedTargetsRequest(field_set.address for field_set in request.field_sets),
96	**implicitly(),
97	)
98	coarsened_targets_by_address = coarsened_targets.by_address()	2✔
99
100	return Partitions(	2✔
101	Partition(
102	tuple(field_sets),
103	PartitionMetadata(
104	CoarsenedTargets(
105	coarsened_targets_by_address[field_set.address] for field_set in field_sets
106	),
107	resolve if len(python_setup.resolves) > 1 else None,
108	InterpreterConstraints.merge((interpreter_constraints, first_party_ics)),
109	),
110	)
111	for (
112	resolve,
113	interpreter_constraints,
114	), field_sets in resolve_and_interpreter_constraints_to_field_sets.items()
115	)
116
117
118	@rule(desc="Lint using Pylint", level=LogLevel.DEBUG)	2✔
119	async def run_pylint(	2✔
120	request: PylintRequest.Batch[PylintFieldSet, PartitionMetadata],
121	pylint: Pylint,
122	first_party_plugins: PylintFirstPartyPlugins,
123	pex_environment: PexEnvironment,
124	) -> LintResult:
125	assert request.partition_metadata is not None	2✔
126
127	# The coarsened targets in the incoming request are for all targets in the request's original
128	# partition. Since the core `lint` logic re-batches inputs according to `[lint].batch_size`,
129	# this could be many more targets than are actually needed to lint the specific batch of files
130	# received by this rule. Subset the CTs one more time here to only those that are relevant.
131	all_coarsened_targets_by_address = request.partition_metadata.coarsened_targets.by_address()	2✔
132	coarsened_targets = CoarsenedTargets(	2✔
133	all_coarsened_targets_by_address[field_set.address] for field_set in request.elements
134	)
135	coarsened_closure = tuple(coarsened_targets.closure())	2✔
136
137	requirements_pex_get = create_pex(	2✔
138	**implicitly(
139	RequirementsPexRequest(
140	(target.address for target in coarsened_closure),
141	# NB: These constraints must be identical to the other PEXes. Otherwise, we risk using
142	# a different version for the requirements than the other two PEXes, which can result
143	# in a PEX runtime error about missing dependencies.
144	hardcoded_interpreter_constraints=request.partition_metadata.interpreter_constraints,
145	)
146	)
147	)
148
149	pylint_pex_get = create_pex(	2✔
150	pylint.to_pex_request(
151	interpreter_constraints=request.partition_metadata.interpreter_constraints,
152	extra_requirements=first_party_plugins.requirement_strings,
153	)
154	)
155
156	sources_get = prepare_python_sources(	2✔
157	PythonSourceFilesRequest(coarsened_closure), **implicitly()
158	)
159	# Ensure that the empty report dir exists.
160	report_directory_digest_get = create_digest(CreateDigest([Directory(REPORT_DIR)]))	2✔
161
162	(	2✔
163	pylint_pex,
164	requirements_pex,
165	sources,
166	report_directory,
167	) = await concurrently(
168	pylint_pex_get,
169	requirements_pex_get,
170	sources_get,
171	report_directory_digest_get,
172	)
173
174	pylint_pex_info = await determine_pex_resolve_info(**implicitly({pylint_pex: Pex}))	2✔
175	astroid_info = pylint_pex_info.find("astroid")	2✔
176	# Astroid is a transitive dependency of pylint and should always be available in the pex.
177	assert astroid_info	2✔
178
179	pylint_runner_pex, config_files = await concurrently(	2✔
180	create_venv_pex(
181	VenvPexRequest(
182	PexRequest(
183	output_filename="pylint_runner.pex",
184	interpreter_constraints=request.partition_metadata.interpreter_constraints,
185	main=pylint.main,
186	internal_only=True,
187	pex_path=[pylint_pex, requirements_pex],
188	),
189	pex_environment.in_sandbox(working_directory=None),
190	# Astroid < 2.9.1 had a regression that prevented the use of symlinks:
191	# https://github.com/PyCQA/pylint/issues/1470
192	site_packages_copies=(astroid_info.version < packaging.version.Version("2.9.1")),
193	),
194	**implicitly(),
195	),
196	find_config_file(pylint.config_request(sources.source_files.snapshot.dirs)),
197	)
198
199	pythonpath = list(sources.source_roots)	2✔
200	if first_party_plugins:	2✔
201	pythonpath.append(first_party_plugins.PREFIX)	2✔
202
203	input_digest = await merge_digests(	2✔
204	MergeDigests(
205	(
206	config_files.snapshot.digest,
207	first_party_plugins.sources_digest,
208	sources.source_files.snapshot.digest,
209	report_directory,
210	)
211	)
212	)
213
214	result = await execute_process(	2✔
215	**implicitly(
216	VenvPexProcess(
217	pylint_runner_pex,
218	argv=generate_argv(request.elements, pylint),
219	input_digest=input_digest,
220	output_directories=(REPORT_DIR,),
221	extra_env={"PEX_EXTRA_SYS_PATH": ":".join(pythonpath)},
222	concurrency_available=len(request.elements),
223	description=f"Run Pylint on {pluralize(len(request.elements), 'target')}.",
224	level=LogLevel.DEBUG,
225	)
226	)
227	)
228	report = await remove_prefix(RemovePrefix(result.output_digest, REPORT_DIR))	2✔
229	return LintResult.create(request, result, report=report)	2✔
230
231
232	def rules():	2✔
233	return (	2✔
234	*collect_rules(),
235	*PylintRequest.rules(),
236	*pex_from_targets.rules(),
237	)

pantsbuild / pants / 25259185675

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