26342152999

Committed 23 May 2026 07:59PM UTC coverage: 91.165% (-1.6%) from 92.792%

Build # 26342152999

Build Type

push

github

Committed by

web-flow

Commit Message

Run Linux ARM CI on Depot runners (#23363)

RunsOn is deprecating their v2 stack, and rather than migrate
to v3 we should use the resources kindly donated by Depot.

GitHub also now has Linux ARM runners, should we need them.

Coverage Stats

87305 of 95766 relevant lines covered (91.16%)

3.87 hits per line

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

46.15

/src/python/pants/backend/project_info/dependencies.py

# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
import itertools
import json
from enum import Enum

from pants.engine.addresses import Addresses
from pants.engine.console import Console
from pants.engine.goal import Goal, GoalSubsystem, LineOriented
from pants.engine.internals.graph import resolve_targets, resolve_unexpanded_targets
from pants.engine.internals.graph import transitive_targets as transitive_targets_get
from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly
from pants.engine.target import (
    AlwaysTraverseDeps,
    DependenciesRequest,
    Targets,
    TransitiveTargetsRequest,
)
from pants.engine.target import Dependencies as DependenciesField
from pants.option.option_types import BoolOption, EnumOption


class DependenciesOutputFormat(Enum):
    """Output format for listing dependencies.

    text: List all dependencies as a single list of targets in plain text.
    json: List all dependencies as a mapping `{target: [dependencies]}`.
    """

    text = "text"
    json = "json"


class DependenciesSubsystem(LineOriented, GoalSubsystem):
    name = "dependencies"
    help = "List the dependencies of the input files/targets."

    transitive = BoolOption(
        default=False,
        help="List all transitive dependencies. If unspecified, list direct dependencies only.",
    )
    closed = BoolOption(
        default=False,
        help="Include the input targets in the output, along with the dependencies.",
    )
    format = EnumOption(
        default=DependenciesOutputFormat.text,
        help="Output format for listing dependencies.",
    )


class Dependencies(Goal):
    subsystem_cls = DependenciesSubsystem
    environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY


async def list_dependencies_as_json(
    addresses: Addresses, dependencies_subsystem: DependenciesSubsystem, console: Console
) -> None:
    """Get dependencies for given addresses and list them in the console in JSON."""
    # NB: We must preserve target generators for the roots, i.e. not replace with their
    # generated targets.
    target_roots = await resolve_unexpanded_targets(addresses)
    # NB: When determining dependencies, though, we replace target generators with their
    # generated targets.
    if dependencies_subsystem.transitive:
        transitive_targets_group = await concurrently(
            transitive_targets_get(
                TransitiveTargetsRequest(
                    (address,), should_traverse_deps_predicate=AlwaysTraverseDeps()
                ),
                **implicitly(),
            )
            for address in addresses
        )

        iterated_targets = []
        for idx, transitive_targets in enumerate(transitive_targets_group):
            targets_collection = {
                str(tgt.address)
                for tgt in (
                    transitive_targets.closure
                    if dependencies_subsystem.closed
                    else transitive_targets.dependencies
                )
            }
            iterated_targets.append(sorted(targets_collection))

    else:
        dependencies_per_target_root = await concurrently(
            resolve_targets(
                **implicitly(
                    DependenciesRequest(
                        tgt.get(DependenciesField),
                        should_traverse_deps_predicate=AlwaysTraverseDeps(),
                    )
                )
            )
            for tgt in target_roots
        )

        iterated_targets = []
        for idx, targets in enumerate(dependencies_per_target_root):
            targets_collection = {str(tgt.address) for tgt in targets}
            if dependencies_subsystem.closed:
                targets_collection.add(str(target_roots[idx].address))
            iterated_targets.append(sorted(targets_collection))

    # The assumption is that when iterating the targets and sending dependency requests
    # for them, the lists of dependencies are returned in the very same order.
    mapping = dict(zip([str(tgt.address) for tgt in target_roots], iterated_targets))
    output = json.dumps(mapping, indent=4)

    with dependencies_subsystem.line_oriented(console) as print_stdout:
        print_stdout(output)


async def list_dependencies_as_plain_text(
    addresses: Addresses, dependencies_subsystem: DependenciesSubsystem, console: Console
) -> None:
    """Get dependencies for given addresses and list them in the console as a single list."""
    if dependencies_subsystem.transitive:
        transitive_targets = await transitive_targets_get(
            TransitiveTargetsRequest(
                addresses, should_traverse_deps_predicate=AlwaysTraverseDeps()
            ),
            **implicitly(),
        )
        targets = Targets(transitive_targets.dependencies)
    else:
        # NB: We must preserve target generators for the roots, i.e. not replace with their
        # generated targets.
        target_roots = await resolve_unexpanded_targets(addresses)
        # NB: When determining dependencies, though, we replace target generators with their
        # generated targets.
        dependencies_per_target_root = await concurrently(
            resolve_targets(
                **implicitly(
                    DependenciesRequest(
                        tgt.get(DependenciesField),
                        should_traverse_deps_predicate=AlwaysTraverseDeps(),
                    )
                )
            )
            for tgt in target_roots
        )
        targets = Targets(itertools.chain.from_iterable(dependencies_per_target_root))

    address_strings = {addr.spec for addr in addresses} if dependencies_subsystem.closed else set()
    for tgt in targets:
        address_strings.add(tgt.address.spec)

    with dependencies_subsystem.line_oriented(console) as print_stdout:
        for address in sorted(address_strings):
            print_stdout(address)


@goal_rule
async def dependencies(
    console: Console, addresses: Addresses, dependencies_subsystem: DependenciesSubsystem
) -> Dependencies:
    if DependenciesOutputFormat.text == dependencies_subsystem.format:
        await list_dependencies_as_plain_text(
            addresses=addresses,
            dependencies_subsystem=dependencies_subsystem,
            console=console,
        )

    elif DependenciesOutputFormat.json == dependencies_subsystem.format:
        await list_dependencies_as_json(
            addresses=addresses,
            dependencies_subsystem=dependencies_subsystem,
            console=console,
        )

    return Dependencies(exit_code=0)


def rules():
    return collect_rules()

1	# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	import itertools	11✔
4	import json	11✔
5	from enum import Enum	11✔
6
7	from pants.engine.addresses import Addresses	11✔
8	from pants.engine.console import Console	11✔
9	from pants.engine.goal import Goal, GoalSubsystem, LineOriented	11✔
10	from pants.engine.internals.graph import resolve_targets, resolve_unexpanded_targets	11✔
11	from pants.engine.internals.graph import transitive_targets as transitive_targets_get	11✔
12	from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly	11✔
13	from pants.engine.target import (	11✔
14	AlwaysTraverseDeps,
15	DependenciesRequest,
16	Targets,
17	TransitiveTargetsRequest,
18	)
19	from pants.engine.target import Dependencies as DependenciesField	11✔
20	from pants.option.option_types import BoolOption, EnumOption	11✔
21
22
23	class DependenciesOutputFormat(Enum):	11✔
24	"""Output format for listing dependencies.
25
26	text: List all dependencies as a single list of targets in plain text.
27	json: List all dependencies as a mapping `{target: [dependencies]}`.
28	"""
29
30	text = "text"	11✔
31	json = "json"	11✔
32
33
34	class DependenciesSubsystem(LineOriented, GoalSubsystem):	11✔
35	name = "dependencies"	11✔
36	help = "List the dependencies of the input files/targets."	11✔
37
38	transitive = BoolOption(	11✔
39	default=False,
40	help="List all transitive dependencies. If unspecified, list direct dependencies only.",
41	)
42	closed = BoolOption(	11✔
43	default=False,
44	help="Include the input targets in the output, along with the dependencies.",
45	)
46	format = EnumOption(	11✔
47	default=DependenciesOutputFormat.text,
48	help="Output format for listing dependencies.",
49	)
50
51
52	class Dependencies(Goal):	11✔
53	subsystem_cls = DependenciesSubsystem	11✔
54	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	11✔
55
56
57	async def list_dependencies_as_json(	11✔
58	addresses: Addresses, dependencies_subsystem: DependenciesSubsystem, console: Console
59	) -> None:
60	"""Get dependencies for given addresses and list them in the console in JSON."""
61	# NB: We must preserve target generators for the roots, i.e. not replace with their
62	# generated targets.
63	target_roots = await resolve_unexpanded_targets(addresses)	×
64	# NB: When determining dependencies, though, we replace target generators with their
65	# generated targets.
66	if dependencies_subsystem.transitive:	×
67	transitive_targets_group = await concurrently(	×
68	transitive_targets_get(
69	TransitiveTargetsRequest(
70	(address,), should_traverse_deps_predicate=AlwaysTraverseDeps()
71	),
72	**implicitly(),
73	)
74	for address in addresses
75	)
76
77	iterated_targets = []	×
78	for idx, transitive_targets in enumerate(transitive_targets_group):	×
79	targets_collection = {	×
80	str(tgt.address)
81	for tgt in (
82	transitive_targets.closure
83	if dependencies_subsystem.closed
84	else transitive_targets.dependencies
85	)
86	}
87	iterated_targets.append(sorted(targets_collection))	×
88
89	else:
90	dependencies_per_target_root = await concurrently(	×
91	resolve_targets(
92	**implicitly(
93	DependenciesRequest(
94	tgt.get(DependenciesField),
95	should_traverse_deps_predicate=AlwaysTraverseDeps(),
96	)
97	)
98	)
99	for tgt in target_roots
100	)
101
102	iterated_targets = []	×
103	for idx, targets in enumerate(dependencies_per_target_root):	×
104	targets_collection = {str(tgt.address) for tgt in targets}	×
105	if dependencies_subsystem.closed:	×
106	targets_collection.add(str(target_roots[idx].address))	×
107	iterated_targets.append(sorted(targets_collection))	×
108
109	# The assumption is that when iterating the targets and sending dependency requests
110	# for them, the lists of dependencies are returned in the very same order.
111	mapping = dict(zip([str(tgt.address) for tgt in target_roots], iterated_targets))	×
112	output = json.dumps(mapping, indent=4)	×
113
114	with dependencies_subsystem.line_oriented(console) as print_stdout:	×
115	print_stdout(output)	×
116
117
118	async def list_dependencies_as_plain_text(	11✔
119	addresses: Addresses, dependencies_subsystem: DependenciesSubsystem, console: Console
120	) -> None:
121	"""Get dependencies for given addresses and list them in the console as a single list."""
122	if dependencies_subsystem.transitive:	×
123	transitive_targets = await transitive_targets_get(	×
124	TransitiveTargetsRequest(
125	addresses, should_traverse_deps_predicate=AlwaysTraverseDeps()
126	),
127	**implicitly(),
128	)
129	targets = Targets(transitive_targets.dependencies)	×
130	else:
131	# NB: We must preserve target generators for the roots, i.e. not replace with their
132	# generated targets.
133	target_roots = await resolve_unexpanded_targets(addresses)	×
134	# NB: When determining dependencies, though, we replace target generators with their
135	# generated targets.
136	dependencies_per_target_root = await concurrently(	×
137	resolve_targets(
138	**implicitly(
139	DependenciesRequest(
140	tgt.get(DependenciesField),
141	should_traverse_deps_predicate=AlwaysTraverseDeps(),
142	)
143	)
144	)
145	for tgt in target_roots
146	)
147	targets = Targets(itertools.chain.from_iterable(dependencies_per_target_root))	×
148
149	address_strings = {addr.spec for addr in addresses} if dependencies_subsystem.closed else set()	×
150	for tgt in targets:	×
151	address_strings.add(tgt.address.spec)	×
152
153	with dependencies_subsystem.line_oriented(console) as print_stdout:	×
154	for address in sorted(address_strings):	×
155	print_stdout(address)	×
156
157
158	@goal_rule	11✔
159	async def dependencies(	11✔
160	console: Console, addresses: Addresses, dependencies_subsystem: DependenciesSubsystem
161	) -> Dependencies:
162	if DependenciesOutputFormat.text == dependencies_subsystem.format:	×
163	await list_dependencies_as_plain_text(	×
164	addresses=addresses,
165	dependencies_subsystem=dependencies_subsystem,
166	console=console,
167	)
168
169	elif DependenciesOutputFormat.json == dependencies_subsystem.format:	×
170	await list_dependencies_as_json(	×
171	addresses=addresses,
172	dependencies_subsystem=dependencies_subsystem,
173	console=console,
174	)
175
176	return Dependencies(exit_code=0)	×
177
178
179	def rules():	11✔
180	return collect_rules()	11✔

pantsbuild / pants / 26342152999

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