19572556425

Committed 21 Nov 2025 01:50PM UTC coverage: 80.295% (+0.007%) from 80.288%

Build # 19572556425

Build Type

Pull #22906

github

Committed by

web-flow

Commit Message

Merge c59ca89b1 into 70dc9fe34

Pull Request Pull Request #22906: Update Coursier default version to v2.1.24

Run Details

3 of 3 new or added lines in 2 files covered. (100.0%)

294 existing lines in 12 files now uncovered.

78385 of 97621 relevant lines covered (80.3%)

3.36 hits per line

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

46.97

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

pantsbuild / pants / 19572556425

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