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

0.0

/src/python/pants/backend/python/goals/debug_goals.py

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

from __future__ import annotations

import json
from dataclasses import dataclass
from enum import Enum
from typing import Any

from pants.backend.project_info.peek import _PeekJsonEncoder
from pants.backend.python.dependency_inference.module_mapper import ResolveName
from pants.backend.python.dependency_inference.parse_python_dependencies import (
    ParsedPythonImportInfo,
    PythonFileDependencies,
)
from pants.backend.python.dependency_inference.rules import (
    ImportResolveResult,
    PythonImportDependenciesInferenceFieldSet,
    ResolvedParsedPythonDependencies,
    ResolvedParsedPythonDependenciesRequest,
    UnownedImportsPossibleOwners,
    UnownedImportsPossibleOwnersRequest,
    _collect_imports_info,
    _exec_parse_deps,
    _find_other_owners_for_unowned_imports,
    import_rules,
    resolve_parsed_dependencies,
)
from pants.backend.python.goals.run_python_source import PythonSourceFieldSet
from pants.backend.python.subsystems.setup import PythonSetup
from pants.build_graph.address import Address
from pants.engine.console import Console
from pants.engine.goal import Goal, GoalSubsystem
from pants.engine.internals.selectors import concurrently
from pants.engine.rules import collect_rules, goal_rule, implicitly, rule
from pants.engine.target import Targets
from pants.option.option_types import EnumOption
from pants.util.strutil import softwrap


class AnalysisFlavor(Enum):
    raw_dependency_inference = "raw_dependency_inference"
    dependency_inference = "dependency_inference"


class DumpPythonSourceAnalysisSubsystem(GoalSubsystem):
    name = "python-dump-source-analysis"
    help = "Dump source analysis for python_source targets."

    flavor = EnumOption(
        "--analysis-flavor",
        default=AnalysisFlavor.dependency_inference,
        help=softwrap(
            f"""\
            The type of information that should be returned.\n
            * `{AnalysisFlavor.dependency_inference.value}`: The results of dependency inference, for every detected import in every file.\n
            * `{AnalysisFlavor.raw_dependency_inference.value}`: The raw intermediate results of the dependency inference process,
            at every stage they're available.
            Potentially useful for debugging the dependency inference process.\n
            """
        ),
    )


class DumpPythonSourceAnalysis(Goal):
    subsystem_cls = DumpPythonSourceAnalysisSubsystem
    environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY  # TODO(#17129) — Migrate this.


@dataclass(frozen=True)
class PythonSourceAnalysis:
    """Information on the inferred imports for a Python file, including all raw intermediate
    results."""

    fs: PythonImportDependenciesInferenceFieldSet
    identified: PythonFileDependencies
    resolved: ResolvedParsedPythonDependencies
    possible_owners: UnownedImportsPossibleOwners


@rule
async def dump_python_source_analysis_single(
    fs: PythonImportDependenciesInferenceFieldSet,
    python_setup: PythonSetup,
) -> PythonSourceAnalysis:
    """Infer the dependencies for a single python fieldset, keeping all the intermediate results."""

    parsed_dependencies = await _exec_parse_deps(fs, python_setup)

    resolve = fs.resolve.normalized_value(python_setup)

    resolved_dependencies = await resolve_parsed_dependencies(
        ResolvedParsedPythonDependenciesRequest(fs, parsed_dependencies, resolve), **implicitly()
    )

    import_deps, unowned_imports = _collect_imports_info(resolved_dependencies.resolve_results)

    imports_to_other_owners = await _find_other_owners_for_unowned_imports(
        UnownedImportsPossibleOwnersRequest(unowned_imports, resolve),
    )

    return PythonSourceAnalysis(
        fs, parsed_dependencies, resolved_dependencies, imports_to_other_owners
    )


@dataclass(frozen=True)
class ImportAnalysis:
    """Information on the inferred imports for a Python file."""

    name: str
    reference: ParsedPythonImportInfo | str
    resolved: ImportResolveResult
    possible_resolve: list[tuple[Address, ResolveName]] | None


@dataclass(frozen=True)
class CollectedImportAnalysis:
    """Collected information on all Python files."""

    imports: list[ImportAnalysis]
    assets: list[ImportAnalysis]


def collect_analysis(raw: PythonSourceAnalysis) -> CollectedImportAnalysis:
    """Collect raw analysis and present it in a helpful per-import format."""
    imports = []

    resolved_results = raw.resolved.resolve_results

    for name, info in raw.identified.imports.items():
        possible_resolve = raw.possible_owners.value.get(name)

        imports.append(
            ImportAnalysis(
                name=name,
                reference=info,
                resolved=resolved_results[name],
                possible_resolve=possible_resolve,
            )
        )

    assets = []
    resolved_assets = raw.resolved.assets

    for name in raw.identified.assets:
        possible_resolve = raw.possible_owners.value.get(name)

        assets.append(
            ImportAnalysis(
                name=name,
                reference=name,  # currently assets don't keep track of their line numbers
                resolved=resolved_assets[name],
                possible_resolve=possible_resolve,
            )
        )

    return CollectedImportAnalysis(imports, assets)


@goal_rule
async def dump_python_source_analysis(
    request: DumpPythonSourceAnalysisSubsystem,
    targets: Targets,
    console: Console,
) -> DumpPythonSourceAnalysis:
    source_field_sets = [
        PythonImportDependenciesInferenceFieldSet.create(tgt)
        for tgt in targets
        if PythonSourceFieldSet.is_applicable(tgt)
    ]

    source_analysis = await concurrently(
        dump_python_source_analysis_single(fs, **implicitly()) for fs in source_field_sets
    )

    output: Any
    if request.flavor == AnalysisFlavor.raw_dependency_inference:
        output = source_analysis
    else:
        output = {str(a.fs.address): collect_analysis(a) for a in source_analysis}

    console.print_stdout(json.dumps(output, cls=_PeekJsonEncoder))
    return DumpPythonSourceAnalysis(exit_code=0)


def rules():
    return [
        *import_rules(),
        *collect_rules(),
    ]

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	×
5
6	import json	×
7	from dataclasses import dataclass	×
8	from enum import Enum	×
9	from typing import Any	×
10
11	from pants.backend.project_info.peek import _PeekJsonEncoder	×
12	from pants.backend.python.dependency_inference.module_mapper import ResolveName	×
13	from pants.backend.python.dependency_inference.parse_python_dependencies import (	×
14	ParsedPythonImportInfo,
15	PythonFileDependencies,
16	)
17	from pants.backend.python.dependency_inference.rules import (	×
18	ImportResolveResult,
19	PythonImportDependenciesInferenceFieldSet,
20	ResolvedParsedPythonDependencies,
21	ResolvedParsedPythonDependenciesRequest,
22	UnownedImportsPossibleOwners,
23	UnownedImportsPossibleOwnersRequest,
24	_collect_imports_info,
25	_exec_parse_deps,
26	_find_other_owners_for_unowned_imports,
27	import_rules,
28	resolve_parsed_dependencies,
29	)
30	from pants.backend.python.goals.run_python_source import PythonSourceFieldSet	×
31	from pants.backend.python.subsystems.setup import PythonSetup	×
32	from pants.build_graph.address import Address	×
33	from pants.engine.console import Console	×
34	from pants.engine.goal import Goal, GoalSubsystem	×
35	from pants.engine.internals.selectors import concurrently	×
36	from pants.engine.rules import collect_rules, goal_rule, implicitly, rule	×
37	from pants.engine.target import Targets	×
38	from pants.option.option_types import EnumOption	×
39	from pants.util.strutil import softwrap	×
40
41
42	class AnalysisFlavor(Enum):	×
43	raw_dependency_inference = "raw_dependency_inference"	×
44	dependency_inference = "dependency_inference"	×
45
46
47	class DumpPythonSourceAnalysisSubsystem(GoalSubsystem):	×
48	name = "python-dump-source-analysis"	×
49	help = "Dump source analysis for python_source targets."	×
50
51	flavor = EnumOption(	×
52	"--analysis-flavor",
53	default=AnalysisFlavor.dependency_inference,
54	help=softwrap(
55	f"""\
56	The type of information that should be returned.\n
57	* `{AnalysisFlavor.dependency_inference.value}`: The results of dependency inference, for every detected import in every file.\n
58	* `{AnalysisFlavor.raw_dependency_inference.value}`: The raw intermediate results of the dependency inference process,
59	at every stage they're available.
60	Potentially useful for debugging the dependency inference process.\n
61	"""
62	),
63	)
64
65
66	class DumpPythonSourceAnalysis(Goal):	×
67	subsystem_cls = DumpPythonSourceAnalysisSubsystem	×
68	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY # TODO(#17129) — Migrate this.	×
69
70
71	@dataclass(frozen=True)	×
72	class PythonSourceAnalysis:	×
73	"""Information on the inferred imports for a Python file, including all raw intermediate
74	results."""
75
76	fs: PythonImportDependenciesInferenceFieldSet	×
77	identified: PythonFileDependencies	×
78	resolved: ResolvedParsedPythonDependencies	×
79	possible_owners: UnownedImportsPossibleOwners	×
80
81
82	@rule	×
83	async def dump_python_source_analysis_single(	×
84	fs: PythonImportDependenciesInferenceFieldSet,
85	python_setup: PythonSetup,
86	) -> PythonSourceAnalysis:
87	"""Infer the dependencies for a single python fieldset, keeping all the intermediate results."""
88
89	parsed_dependencies = await _exec_parse_deps(fs, python_setup)	×
90
91	resolve = fs.resolve.normalized_value(python_setup)	×
92
93	resolved_dependencies = await resolve_parsed_dependencies(	×
94	ResolvedParsedPythonDependenciesRequest(fs, parsed_dependencies, resolve), **implicitly()
95	)
96
97	import_deps, unowned_imports = _collect_imports_info(resolved_dependencies.resolve_results)	×
98
99	imports_to_other_owners = await _find_other_owners_for_unowned_imports(	×
100	UnownedImportsPossibleOwnersRequest(unowned_imports, resolve),
101	)
102
103	return PythonSourceAnalysis(	×
104	fs, parsed_dependencies, resolved_dependencies, imports_to_other_owners
105	)
106
107
108	@dataclass(frozen=True)	×
109	class ImportAnalysis:	×
110	"""Information on the inferred imports for a Python file."""
111
112	name: str	×
113	reference: ParsedPythonImportInfo \| str	×
114	resolved: ImportResolveResult	×
115	possible_resolve: list[tuple[Address, ResolveName]] \| None	×
116
117
118	@dataclass(frozen=True)	×
119	class CollectedImportAnalysis:	×
120	"""Collected information on all Python files."""
121
122	imports: list[ImportAnalysis]	×
123	assets: list[ImportAnalysis]	×
124
125
126	def collect_analysis(raw: PythonSourceAnalysis) -> CollectedImportAnalysis:	×
127	"""Collect raw analysis and present it in a helpful per-import format."""
128	imports = []	×
129
130	resolved_results = raw.resolved.resolve_results	×
131
132	for name, info in raw.identified.imports.items():	×
133	possible_resolve = raw.possible_owners.value.get(name)	×
134
135	imports.append(	×
136	ImportAnalysis(
137	name=name,
138	reference=info,
139	resolved=resolved_results[name],
140	possible_resolve=possible_resolve,
141	)
142	)
143
144	assets = []	×
145	resolved_assets = raw.resolved.assets	×
146
147	for name in raw.identified.assets:	×
148	possible_resolve = raw.possible_owners.value.get(name)	×
149
150	assets.append(	×
151	ImportAnalysis(
152	name=name,
153	reference=name, # currently assets don't keep track of their line numbers
154	resolved=resolved_assets[name],
155	possible_resolve=possible_resolve,
156	)
157	)
158
159	return CollectedImportAnalysis(imports, assets)	×
160
161
162	@goal_rule	×
163	async def dump_python_source_analysis(	×
164	request: DumpPythonSourceAnalysisSubsystem,
165	targets: Targets,
166	console: Console,
167	) -> DumpPythonSourceAnalysis:
168	source_field_sets = [	×
169	PythonImportDependenciesInferenceFieldSet.create(tgt)
170	for tgt in targets
171	if PythonSourceFieldSet.is_applicable(tgt)
172	]
173
174	source_analysis = await concurrently(	×
175	dump_python_source_analysis_single(fs, **implicitly()) for fs in source_field_sets
176	)
177
178	output: Any
179	if request.flavor == AnalysisFlavor.raw_dependency_inference:	×
180	output = source_analysis	×
181	else:
182	output = {str(a.fs.address): collect_analysis(a) for a in source_analysis}	×
183
184	console.print_stdout(json.dumps(output, cls=_PeekJsonEncoder))	×
185	return DumpPythonSourceAnalysis(exit_code=0)	×
186
187
188	def rules():	×
189	return [	×
190	*import_rules(),
191	*collect_rules(),
192	]

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