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

92.13

/src/python/pants/jvm/dependency_inference/artifact_mapper.py

# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

from collections import defaultdict
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from typing import Any, DefaultDict

from pants.backend.java.subsystems.java_infer import JavaInferSubsystem
from pants.build_graph.address import Address
from pants.engine.rules import collect_rules, rule
from pants.engine.target import AllTargets, Targets
from pants.jvm.dependency_inference.jvm_artifact_mappings import JVM_ARTIFACT_MAPPINGS
from pants.jvm.resolve.common import ArtifactRequirement
from pants.jvm.resolve.coordinate import Coordinate
from pants.jvm.subsystems import JvmSubsystem
from pants.jvm.target_types import (
    JvmArtifactArtifactField,
    JvmArtifactGroupField,
    JvmArtifactPackagesField,
    JvmProvidesTypesField,
    JvmResolveField,
)
from pants.util.docutil import bin_name
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet, OrderedSet

_ResolveName = str


@dataclass(frozen=True)
class UnversionedCoordinate:
    group: str
    artifact: str

    @classmethod
    def from_coord_str(cls, coord: str) -> UnversionedCoordinate:
        coordinate_parts = coord.split(":")
        if len(coordinate_parts) != 2:
            raise ValueError(f"Invalid coordinate specifier: {coord}")
        return UnversionedCoordinate(group=coordinate_parts[0], artifact=coordinate_parts[1])


class AvailableThirdPartyArtifacts(
    FrozenDict[
        tuple[_ResolveName, UnversionedCoordinate], tuple[tuple[Address, ...], tuple[str, ...]]
    ]
):
    """Maps coordinates and resolve names to target `Address`es and declared packages."""


# A namespace for a symbol which defines the scope for name collisions and ambiguity. For example:
# if a JVM language allows the same symbol to be declared in an unambiguous way in multiple files
# (such as Scala `package objects` declaring a type alias for a class/object in another file) it
# can use a non-default namespace name of its choice.
SymbolNamespace = str


DEFAULT_SYMBOL_NAMESPACE: SymbolNamespace = "default"


class MutableTrieNode:
    __slots__ = [
        "children",
        "recursive",
        "addresses",
        "first_party",
    ]  # don't use a `dict` to store attrs

    def __init__(self) -> None:
        self.children: dict[str, MutableTrieNode] = {}
        self.recursive: bool = False
        self.addresses: dict[SymbolNamespace, OrderedSet[Address]] = defaultdict(OrderedSet)
        self.first_party: bool = False

    def _ensure_child(self, name: str) -> MutableTrieNode:
        if name in self.children:
            return self.children[name]
        node = MutableTrieNode()
        self.children[name] = node
        return node

    def insert(
        self,
        symbol: str,
        addresses: Iterable[Address],
        *,
        first_party: bool,
        namespace: SymbolNamespace = DEFAULT_SYMBOL_NAMESPACE,
        recursive: bool = False,
    ) -> None:
        imp_parts = symbol.split(".")
        current_node = self
        for imp_part in imp_parts:
            child_node = current_node._ensure_child(imp_part)
            current_node = child_node

        current_node.addresses[namespace].update(addresses)
        current_node.first_party = first_party
        current_node.recursive = recursive

    def frozen(self) -> FrozenTrieNode:
        return FrozenTrieNode(self)


FrozenTrieNodeItem = tuple[str, bool, FrozenDict[SymbolNamespace, FrozenOrderedSet[Address]], bool]


@dataclass(frozen=True)
class FrozenTrieNode:
    __slots__ = [
        "_children",
        "_recursive",
        "_addresses",
        "_first_party",
    ]  # don't use a `dict` to store attrs (speeds up attr access significantly)

    _children: FrozenDict[str, FrozenTrieNode]
    _recursive: bool
    _addresses: FrozenDict[SymbolNamespace, FrozenOrderedSet[Address]]
    _first_party: bool

    def __init__(self, node: MutableTrieNode) -> None:
        children = {}
        for key, child in node.children.items():
            children[key] = FrozenTrieNode(child)

        object.__setattr__(self, "_children", FrozenDict(children))
        object.__setattr__(self, "_recursive", node.recursive)
        object.__setattr__(
            self,
            "_addresses",
            FrozenDict(
                {ns: FrozenOrderedSet(addresses) for ns, addresses in node.addresses.items()}
            ),
        )
        object.__setattr__(self, "_first_party", node.first_party)

    def find_child(self, name: str) -> FrozenTrieNode | None:
        return self._children.get(name)

    @property
    def recursive(self) -> bool:
        return self._recursive

    @property
    def first_party(self) -> bool:
        return self._first_party

    def addresses_for_symbol(
        self, symbol: str
    ) -> FrozenDict[SymbolNamespace, FrozenOrderedSet[Address]]:
        current_node = self
        imp_parts = symbol.split(".")

        found_nodes = []
        for imp_part in imp_parts:
            child_node_opt = current_node.find_child(imp_part)
            if not child_node_opt:
                break
            found_nodes.append(child_node_opt)
            current_node = child_node_opt

        if not found_nodes:
            return FrozenDict()

        # If the length of the found nodes equals the number of parts of the package path, then
        # there is an exact match.
        if len(found_nodes) == len(imp_parts):
            return found_nodes[-1].addresses

        # Otherwise, check for the first found node (in reverse order) to match recursively, and
        # use its coordinate.
        for found_node in reversed(found_nodes):
            if found_node.recursive:
                return found_node.addresses

        # Nothing matched so return no match.
        return FrozenDict()

    @property
    def addresses(self) -> FrozenDict[SymbolNamespace, FrozenOrderedSet[Address]]:
        return self._addresses

    @classmethod
    def merge(cls, nodes: Iterable[FrozenTrieNode]) -> FrozenTrieNode:
        """Merges the given `FrozenTrieNode` instances.

        TODO: This is currently implemented as merging-from-scratch, but could be trie-aware.
        """
        result = MutableTrieNode()
        for node in nodes:
            for symbol, recursive, address_namespaces, first_party in node:
                for namespace, addresses in address_namespaces.items():
                    result.insert(
                        symbol,
                        addresses,
                        recursive=recursive,
                        first_party=first_party,
                        namespace=namespace,
                    )
        return FrozenTrieNode(result)

    def to_json_dict(self) -> dict[str, Any]:
        return {
            "children": {name: child.to_json_dict() for name, child in self._children.items()},
            "addresses": {
                ns: [str(a) for a in addresses] for ns, addresses in self._addresses.items()
            },
            "recursive": self._recursive,
            "first_party": self._first_party,
        }

    def _iter_helper(self, symbol: list[str]) -> Iterator[FrozenTrieNodeItem]:
        if symbol and self._addresses:
            yield (".".join(symbol), self._recursive, self._addresses, self._first_party)
        for name, child in self._children.items():
            symbol.append(name)
            yield from child._iter_helper(symbol)
            symbol.pop()

    def __iter__(self) -> Iterator[FrozenTrieNodeItem]:
        """Iterates through all nodes in the trie.

        TODO: This is primarily used in `FrozenTrieNode.merge`: if that method switches to
        trie-aware merging, this should likely be removed.
        """
        yield from self._iter_helper([])

    def __hash__(self) -> int:
        return hash((self._children, self._recursive, self._addresses))

    def __eq__(self, other: Any) -> bool:
        if not isinstance(other, FrozenTrieNode):
            return False
        return (
            self.recursive == other.recursive
            and self.first_party == other.first_party
            and self.addresses == other.addresses
            and self._children == other._children
        )

    def __repr__(self):
        return f"FrozenTrieNode(children={repr(self._children)}, recursive={self._recursive}, addresses={self._addresses}, first_party={self._first_party})"


class AllJvmArtifactTargets(Targets):
    pass


class AllJvmTypeProvidingTargets(Targets):
    pass


@rule(desc="Find all jvm_artifact targets in project", level=LogLevel.DEBUG)
async def find_all_jvm_artifact_targets(targets: AllTargets) -> AllJvmArtifactTargets:
    return AllJvmArtifactTargets(
        tgt for tgt in targets if tgt.has_fields((JvmArtifactGroupField, JvmArtifactArtifactField))
    )


@rule(desc="Find all targets with experimental_provides fields in project", level=LogLevel.DEBUG)
async def find_all_jvm_provides_fields(targets: AllTargets) -> AllJvmTypeProvidingTargets:
    return AllJvmTypeProvidingTargets(
        tgt
        for tgt in targets
        if tgt.has_field(JvmProvidesTypesField) and tgt[JvmProvidesTypesField].value is not None
    )


class ThirdPartySymbolMapping(FrozenDict[_ResolveName, FrozenTrieNode]):
    """The third party symbols provided by all `jvm_artifact` targets."""


@rule
async def find_available_third_party_artifacts(
    all_jvm_artifact_tgts: AllJvmArtifactTargets, jvm: JvmSubsystem
) -> AvailableThirdPartyArtifacts:
    address_mapping: DefaultDict[
        tuple[_ResolveName, UnversionedCoordinate], OrderedSet[Address]
    ] = defaultdict(OrderedSet)
    package_mapping: DefaultDict[tuple[_ResolveName, UnversionedCoordinate], OrderedSet[str]] = (
        defaultdict(OrderedSet)
    )
    for tgt in all_jvm_artifact_tgts:
        coord = UnversionedCoordinate(
            group=tgt[JvmArtifactGroupField].value, artifact=tgt[JvmArtifactArtifactField].value
        )
        resolve = tgt[JvmResolveField].normalized_value(jvm)
        key = (resolve, coord)
        address_mapping[key].add(tgt.address)
        package_mapping[key].update(tgt[JvmArtifactPackagesField].value or ())

    return AvailableThirdPartyArtifacts(
        {
            key: (tuple(addresses), tuple(package_mapping[key]))
            for key, addresses in address_mapping.items()
        }
    )


@rule
async def compute_java_third_party_symbol_mapping(
    java_infer_subsystem: JavaInferSubsystem,
    available_artifacts: AvailableThirdPartyArtifacts,
    all_jvm_type_providing_tgts: AllJvmTypeProvidingTargets,
) -> ThirdPartySymbolMapping:
    """Implements the mapping logic from the `jvm_artifact` and `java-infer` help."""

    def symbol_from_package_pattern(package_pattern: str) -> tuple[str, bool]:
        wildcard_suffix = ".**"
        if package_pattern.endswith(wildcard_suffix):
            return package_pattern[: -len(wildcard_suffix)], True
        else:
            return package_pattern, False

    # Build a default mapping from coord to package.
    # TODO: Consider inverting the definitions of these mappings.
    default_coords_to_packages: dict[UnversionedCoordinate, OrderedSet[str]] = defaultdict(
        OrderedSet
    )
    for package, unversioned_coord_str in {
        **JVM_ARTIFACT_MAPPINGS,
        **java_infer_subsystem.third_party_import_mapping,
    }.items():
        unversioned_coord = UnversionedCoordinate.from_coord_str(unversioned_coord_str)
        default_coords_to_packages[unversioned_coord].add(package)

    # Build mappings per resolve from packages to addresses.
    mappings: DefaultDict[_ResolveName, MutableTrieNode] = defaultdict(MutableTrieNode)
    for (resolve_name, coord), (addresses, packages) in available_artifacts.items():
        if not packages:
            # If no packages were explicitly defined, fall back to our default mapping.
            packages = tuple(default_coords_to_packages[coord])
        if not packages:
            # Default to exposing the `group` name as a package.
            packages = (f"{coord.group}.**",)
        mapping = mappings[resolve_name]
        for package in packages:
            symbol, recursive = symbol_from_package_pattern(package)
            mapping.insert(symbol, addresses, first_party=False, recursive=recursive)

    # Mark types that have strong first-party declarations as first-party
    for tgt in all_jvm_type_providing_tgts:
        for provides_type in tgt[JvmProvidesTypesField].value or []:
            for mapping in mappings.values():
                mapping.insert(provides_type, [], first_party=True, recursive=False)

    return ThirdPartySymbolMapping(
        FrozenDict(
            (resolve_name, FrozenTrieNode(mapping)) for resolve_name, mapping in mappings.items()
        )
    )


class ConflictingJvmArtifactVersionInResolveError(ValueError):
    def __init__(
        self,
        *,
        subsystem: str,
        requirement_source: str | None = None,
        resolve_name: str,
        required_version: str,
        conflicting_coordinate: Coordinate,
    ) -> None:
        source = f" from {requirement_source}" if requirement_source else ""
        msg = (
            f"The JVM resolve `{resolve_name}` contains a `jvm_artifact` for version {conflicting_coordinate.version} "
            f"of {subsystem}. This conflicts with version {required_version} which is the configured version "
            f"of {subsystem} for this resolve{source}. "
            "Please remove the `jvm_artifact` target with JVM coordinate "
            f"{conflicting_coordinate.to_coord_str()}, then re-run "
            f"`{bin_name()} generate-lockfiles --resolve={resolve_name}`"
        )
        super().__init__(msg)


class MissingRequiredJvmArtifactsInResolve(ValueError):
    def __init__(
        self,
        coordinates: Iterable[Coordinate | UnversionedCoordinate],
        *,
        subsystem: str,
        resolve_name: str,
        target_type: str,
    ) -> None:
        msg = (
            f"The JVM resolve `{resolve_name}` is missing one or more requirements for {subsystem}. "
            f"Since at least one JVM target type in this repository consumes a `{target_type}` target "
            "in this resolve, this resolve must contain `jvm_artifact` targets for each requirement of "
            f"{subsystem}.\n\n"
            "Please add the following `jvm_artifact` target(s) somewhere in the repository and re-run "
            f"`{bin_name()} generate-lockfiles --resolve={resolve_name}`:\n"
        )
        for coordinate in coordinates:
            if isinstance(coordinate, Coordinate):
                msg += (
                    "\njvm_artifact(\n"
                    f'  name="{coordinate.group}_{coordinate.artifact}",\n'
                    f'  group="{coordinate.group}",\n'
                    f'  artifact="{coordinate.artifact}",\n'
                    f'  version="{coordinate.version}",\n'
                    f'  resolve="{resolve_name}",\n'
                    ")\n"
                )
            elif isinstance(coordinate, UnversionedCoordinate):
                msg += (
                    "\njvm_artifact(\n"
                    f'  name="{coordinate.group}_{coordinate.artifact}",\n'
                    f'  group="{coordinate.group}",\n'
                    f'  artifact="{coordinate.artifact}",\n'
                    '  version="<your preferred runtime version>",\n'
                    f'  resolve="{resolve_name}",\n'
                    ")\n"
                )
        super().__init__(msg)


def find_jvm_artifacts_or_raise(
    required_coordinates: Iterable[Coordinate | UnversionedCoordinate],
    resolve: str,
    jvm_artifact_targets: AllJvmArtifactTargets,
    jvm: JvmSubsystem,
    *,
    subsystem: str,
    target_type: str,
    requirement_source: str | None = None,
) -> frozenset[Address]:
    remaining_coordinates: set[Coordinate | UnversionedCoordinate] = set(required_coordinates)

    addresses: set[Address] = set()
    for tgt in jvm_artifact_targets:
        if tgt[JvmResolveField].normalized_value(jvm) != resolve:
            continue

        artifact = ArtifactRequirement.from_jvm_artifact_target(tgt)
        found_coordinates: set[Coordinate | UnversionedCoordinate] = set()
        for coordinate in remaining_coordinates:
            if isinstance(coordinate, Coordinate):
                if (
                    artifact.coordinate.group != coordinate.group
                    or artifact.coordinate.artifact != coordinate.artifact
                ):
                    continue
                if artifact.coordinate.version != coordinate.version:
                    raise ConflictingJvmArtifactVersionInResolveError(
                        subsystem=subsystem,
                        requirement_source=requirement_source,
                        resolve_name=resolve,
                        required_version=coordinate.version,
                        conflicting_coordinate=artifact.coordinate,
                    )
            elif isinstance(coordinate, UnversionedCoordinate):
                if (
                    artifact.coordinate.group != coordinate.group
                    or artifact.coordinate.artifact != coordinate.artifact
                ):
                    continue

            found_coordinates.add(coordinate)

        if found_coordinates:
            remaining_coordinates.difference_update(found_coordinates)
            addresses.add(tgt.address)

    if remaining_coordinates:
        raise MissingRequiredJvmArtifactsInResolve(
            remaining_coordinates,
            subsystem=subsystem,
            resolve_name=resolve,
            target_type=target_type,
        )

    return frozenset(addresses)


def rules():
    return collect_rules()

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	9✔
4
5	from collections import defaultdict	9✔
6	from collections.abc import Iterable, Iterator	9✔
7	from dataclasses import dataclass	9✔
8	from typing import Any, DefaultDict	9✔
9
10	from pants.backend.java.subsystems.java_infer import JavaInferSubsystem	9✔
11	from pants.build_graph.address import Address	9✔
12	from pants.engine.rules import collect_rules, rule	9✔
13	from pants.engine.target import AllTargets, Targets	9✔
14	from pants.jvm.dependency_inference.jvm_artifact_mappings import JVM_ARTIFACT_MAPPINGS	9✔
15	from pants.jvm.resolve.common import ArtifactRequirement	9✔
16	from pants.jvm.resolve.coordinate import Coordinate	9✔
17	from pants.jvm.subsystems import JvmSubsystem	9✔
18	from pants.jvm.target_types import (	9✔
19	JvmArtifactArtifactField,
20	JvmArtifactGroupField,
21	JvmArtifactPackagesField,
22	JvmProvidesTypesField,
23	JvmResolveField,
24	)
25	from pants.util.docutil import bin_name	9✔
26	from pants.util.frozendict import FrozenDict	9✔
27	from pants.util.logging import LogLevel	9✔
28	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	9✔
29
30	_ResolveName = str	9✔
31
32
33	@dataclass(frozen=True)	9✔
34	class UnversionedCoordinate:	9✔
35	group: str	9✔
36	artifact: str	9✔
37
38	@classmethod	9✔
39	def from_coord_str(cls, coord: str) -> UnversionedCoordinate:	9✔
40	coordinate_parts = coord.split(":")	8✔
41	if len(coordinate_parts) != 2:	8✔
42	raise ValueError(f"Invalid coordinate specifier: {coord}")	×
43	return UnversionedCoordinate(group=coordinate_parts[0], artifact=coordinate_parts[1])	8✔
44
45
46	class AvailableThirdPartyArtifacts(	9✔
47	FrozenDict[
48	tuple[_ResolveName, UnversionedCoordinate], tuple[tuple[Address, ...], tuple[str, ...]]
49	]
50	):
51	"""Maps coordinates and resolve names to target `Address`es and declared packages."""
52
53
54	# A namespace for a symbol which defines the scope for name collisions and ambiguity. For example:
55	# if a JVM language allows the same symbol to be declared in an unambiguous way in multiple files
56	# (such as Scala `package objects` declaring a type alias for a class/object in another file) it
57	# can use a non-default namespace name of its choice.
58	SymbolNamespace = str	9✔
59
60
61	DEFAULT_SYMBOL_NAMESPACE: SymbolNamespace = "default"	9✔
62
63
64	class MutableTrieNode:	9✔
65	__slots__ = [	9✔
66	"children",
67	"recursive",
68	"addresses",
69	"first_party",
70	] # don't use a `dict` to store attrs
71
72	def __init__(self) -> None:	9✔
73	self.children: dict[str, MutableTrieNode] = {}	8✔
74	self.recursive: bool = False	8✔
75	self.addresses: dict[SymbolNamespace, OrderedSet[Address]] = defaultdict(OrderedSet)	8✔
76	self.first_party: bool = False	8✔
77
78	def _ensure_child(self, name: str) -> MutableTrieNode:	9✔
79	if name in self.children:	8✔
80	return self.children[name]	8✔
81	node = MutableTrieNode()	8✔
82	self.children[name] = node	8✔
83	return node	8✔
84
85	def insert(	9✔
86	self,
87	symbol: str,
88	addresses: Iterable[Address],
89	*,
90	first_party: bool,
91	namespace: SymbolNamespace = DEFAULT_SYMBOL_NAMESPACE,
92	recursive: bool = False,
93	) -> None:
94	imp_parts = symbol.split(".")	8✔
95	current_node = self	8✔
96	for imp_part in imp_parts:	8✔
97	child_node = current_node._ensure_child(imp_part)	8✔
98	current_node = child_node	8✔
99
100	current_node.addresses[namespace].update(addresses)	8✔
101	current_node.first_party = first_party	8✔
102	current_node.recursive = recursive	8✔
103
104	def frozen(self) -> FrozenTrieNode:	9✔
105	return FrozenTrieNode(self)	8✔
106
107
108	FrozenTrieNodeItem = tuple[str, bool, FrozenDict[SymbolNamespace, FrozenOrderedSet[Address]], bool]	9✔
109
110
111	@dataclass(frozen=True)	9✔
112	class FrozenTrieNode:	9✔
113	__slots__ = [	9✔
114	"_children",
115	"_recursive",
116	"_addresses",
117	"_first_party",
118	] # don't use a `dict` to store attrs (speeds up attr access significantly)
119
120	_children: FrozenDict[str, FrozenTrieNode]	9✔
121	_recursive: bool	9✔
122	_addresses: FrozenDict[SymbolNamespace, FrozenOrderedSet[Address]]	9✔
123	_first_party: bool	9✔
124
125	def __init__(self, node: MutableTrieNode) -> None:	9✔
126	children = {}	8✔
127	for key, child in node.children.items():	8✔
128	children[key] = FrozenTrieNode(child)	8✔
129
130	object.__setattr__(self, "_children", FrozenDict(children))	8✔
131	object.__setattr__(self, "_recursive", node.recursive)	8✔
132	object.__setattr__(	8✔
133	self,
134	"_addresses",
135	FrozenDict(
136	{ns: FrozenOrderedSet(addresses) for ns, addresses in node.addresses.items()}
137	),
138	)
139	object.__setattr__(self, "_first_party", node.first_party)	8✔
140
141	def find_child(self, name: str) -> FrozenTrieNode \| None:	9✔
142	return self._children.get(name)	8✔
143
144	@property	9✔
145	def recursive(self) -> bool:	9✔
146	return self._recursive	8✔
147
148	@property	9✔
149	def first_party(self) -> bool:	9✔
150	return self._first_party	1✔
151
152	def addresses_for_symbol(	9✔
153	self, symbol: str
154	) -> FrozenDict[SymbolNamespace, FrozenOrderedSet[Address]]:
155	current_node = self	8✔
156	imp_parts = symbol.split(".")	8✔
157
158	found_nodes = []	8✔
159	for imp_part in imp_parts:	8✔
160	child_node_opt = current_node.find_child(imp_part)	8✔
161	if not child_node_opt:	8✔
162	break	8✔
163	found_nodes.append(child_node_opt)	8✔
164	current_node = child_node_opt	8✔
165
166	if not found_nodes:	8✔
167	return FrozenDict()	6✔
168
169	# If the length of the found nodes equals the number of parts of the package path, then
170	# there is an exact match.
171	if len(found_nodes) == len(imp_parts):	8✔
172	return found_nodes[-1].addresses	7✔
173
174	# Otherwise, check for the first found node (in reverse order) to match recursively, and
175	# use its coordinate.
176	for found_node in reversed(found_nodes):	8✔
177	if found_node.recursive:	8✔
178	return found_node.addresses	7✔
179
180	# Nothing matched so return no match.
181	return FrozenDict()	7✔
182
183	@property	9✔
184	def addresses(self) -> FrozenDict[SymbolNamespace, FrozenOrderedSet[Address]]:	9✔
185	return self._addresses	8✔
186
187	@classmethod	9✔
188	def merge(cls, nodes: Iterable[FrozenTrieNode]) -> FrozenTrieNode:	9✔
189	"""Merges the given `FrozenTrieNode` instances.
190
191	TODO: This is currently implemented as merging-from-scratch, but could be trie-aware.
192	"""
193	result = MutableTrieNode()	8✔
194	for node in nodes:	8✔
195	for symbol, recursive, address_namespaces, first_party in node:	8✔
196	for namespace, addresses in address_namespaces.items():	8✔
197	result.insert(	8✔
198	symbol,
199	addresses,
200	recursive=recursive,
201	first_party=first_party,
202	namespace=namespace,
203	)
204	return FrozenTrieNode(result)	8✔
205
206	def to_json_dict(self) -> dict[str, Any]:	9✔
207	return {	×
208	"children": {name: child.to_json_dict() for name, child in self._children.items()},
209	"addresses": {
210	ns: [str(a) for a in addresses] for ns, addresses in self._addresses.items()
211	},
212	"recursive": self._recursive,
213	"first_party": self._first_party,
214	}
215
216	def _iter_helper(self, symbol: list[str]) -> Iterator[FrozenTrieNodeItem]:	9✔
217	if symbol and self._addresses:	8✔
218	yield (".".join(symbol), self._recursive, self._addresses, self._first_party)	8✔
219	for name, child in self._children.items():	8✔
220	symbol.append(name)	8✔
221	yield from child._iter_helper(symbol)	8✔
222	symbol.pop()	8✔
223
224	def __iter__(self) -> Iterator[FrozenTrieNodeItem]:	9✔
225	"""Iterates through all nodes in the trie.
226
227	TODO: This is primarily used in `FrozenTrieNode.merge`: if that method switches to
228	trie-aware merging, this should likely be removed.
229	"""
230	yield from self._iter_helper([])	8✔
231
232	def __hash__(self) -> int:	9✔
233	return hash((self._children, self._recursive, self._addresses))	8✔
234
235	def __eq__(self, other: Any) -> bool:	9✔
236	if not isinstance(other, FrozenTrieNode):	1✔
237	return False	×
238	return (	1✔
239	self.recursive == other.recursive
240	and self.first_party == other.first_party
241	and self.addresses == other.addresses
242	and self._children == other._children
243	)
244
245	def __repr__(self):	9✔
246	return f"FrozenTrieNode(children={repr(self._children)}, recursive={self._recursive}, addresses={self._addresses}, first_party={self._first_party})"	×
247
248
249	class AllJvmArtifactTargets(Targets):	9✔
250	pass	9✔
251
252
253	class AllJvmTypeProvidingTargets(Targets):	9✔
254	pass	9✔
255
256
257	@rule(desc="Find all jvm_artifact targets in project", level=LogLevel.DEBUG)	9✔
258	async def find_all_jvm_artifact_targets(targets: AllTargets) -> AllJvmArtifactTargets:	9✔
259	return AllJvmArtifactTargets(	8✔
260	tgt for tgt in targets if tgt.has_fields((JvmArtifactGroupField, JvmArtifactArtifactField))
261	)
262
263
264	@rule(desc="Find all targets with experimental_provides fields in project", level=LogLevel.DEBUG)	9✔
265	async def find_all_jvm_provides_fields(targets: AllTargets) -> AllJvmTypeProvidingTargets:	9✔
266	return AllJvmTypeProvidingTargets(	8✔
267	tgt
268	for tgt in targets
269	if tgt.has_field(JvmProvidesTypesField) and tgt[JvmProvidesTypesField].value is not None
270	)
271
272
273	class ThirdPartySymbolMapping(FrozenDict[_ResolveName, FrozenTrieNode]):	9✔
274	"""The third party symbols provided by all `jvm_artifact` targets."""
275
276
277	@rule	9✔
278	async def find_available_third_party_artifacts(	9✔
279	all_jvm_artifact_tgts: AllJvmArtifactTargets, jvm: JvmSubsystem
280	) -> AvailableThirdPartyArtifacts:
281	address_mapping: DefaultDict[	8✔
282	tuple[_ResolveName, UnversionedCoordinate], OrderedSet[Address]
283	] = defaultdict(OrderedSet)
284	package_mapping: DefaultDict[tuple[_ResolveName, UnversionedCoordinate], OrderedSet[str]] = (	8✔
285	defaultdict(OrderedSet)
286	)
287	for tgt in all_jvm_artifact_tgts:	8✔
288	coord = UnversionedCoordinate(	8✔
289	group=tgt[JvmArtifactGroupField].value, artifact=tgt[JvmArtifactArtifactField].value
290	)
291	resolve = tgt[JvmResolveField].normalized_value(jvm)	8✔
292	key = (resolve, coord)	8✔
293	address_mapping[key].add(tgt.address)	8✔
294	package_mapping[key].update(tgt[JvmArtifactPackagesField].value or ())	8✔
295
296	return AvailableThirdPartyArtifacts(	8✔
297	{
298	key: (tuple(addresses), tuple(package_mapping[key]))
299	for key, addresses in address_mapping.items()
300	}
301	)
302
303
304	@rule	9✔
305	async def compute_java_third_party_symbol_mapping(	9✔
306	java_infer_subsystem: JavaInferSubsystem,
307	available_artifacts: AvailableThirdPartyArtifacts,
308	all_jvm_type_providing_tgts: AllJvmTypeProvidingTargets,
309	) -> ThirdPartySymbolMapping:
310	"""Implements the mapping logic from the `jvm_artifact` and `java-infer` help."""
311
312	def symbol_from_package_pattern(package_pattern: str) -> tuple[str, bool]:	8✔
313	wildcard_suffix = ".**"	8✔
314	if package_pattern.endswith(wildcard_suffix):	8✔
315	return package_pattern[: -len(wildcard_suffix)], True	8✔
316	else:
317	return package_pattern, False	1✔
318
319	# Build a default mapping from coord to package.
320	# TODO: Consider inverting the definitions of these mappings.
321	default_coords_to_packages: dict[UnversionedCoordinate, OrderedSet[str]] = defaultdict(	8✔
322	OrderedSet
323	)
324	for package, unversioned_coord_str in {	8✔
325	**JVM_ARTIFACT_MAPPINGS,
326	**java_infer_subsystem.third_party_import_mapping,
327	}.items():
328	unversioned_coord = UnversionedCoordinate.from_coord_str(unversioned_coord_str)	8✔
329	default_coords_to_packages[unversioned_coord].add(package)	8✔
330
331	# Build mappings per resolve from packages to addresses.
332	mappings: DefaultDict[_ResolveName, MutableTrieNode] = defaultdict(MutableTrieNode)	8✔
333	for (resolve_name, coord), (addresses, packages) in available_artifacts.items():	8✔
334	if not packages:	8✔
335	# If no packages were explicitly defined, fall back to our default mapping.
336	packages = tuple(default_coords_to_packages[coord])	8✔
337	if not packages:	8✔
338	# Default to exposing the `group` name as a package.
339	packages = (f"{coord.group}.**",)	8✔
340	mapping = mappings[resolve_name]	8✔
341	for package in packages:	8✔
342	symbol, recursive = symbol_from_package_pattern(package)	8✔
343	mapping.insert(symbol, addresses, first_party=False, recursive=recursive)	8✔
344
345	# Mark types that have strong first-party declarations as first-party
346	for tgt in all_jvm_type_providing_tgts:	8✔
347	for provides_type in tgt[JvmProvidesTypesField].value or []:	1✔
348	for mapping in mappings.values():	1✔
349	mapping.insert(provides_type, [], first_party=True, recursive=False)	1✔
350
351	return ThirdPartySymbolMapping(	8✔
352	FrozenDict(
353	(resolve_name, FrozenTrieNode(mapping)) for resolve_name, mapping in mappings.items()
354	)
355	)
356
357
358	class ConflictingJvmArtifactVersionInResolveError(ValueError):	9✔
359	def __init__(	9✔
360	self,
361	*,
362	subsystem: str,
363	requirement_source: str \| None = None,
364	resolve_name: str,
365	required_version: str,
366	conflicting_coordinate: Coordinate,
367	) -> None:
368	source = f" from {requirement_source}" if requirement_source else ""	×
369	msg = (	×
370	f"The JVM resolve `{resolve_name}` contains a `jvm_artifact` for version {conflicting_coordinate.version} "
371	f"of {subsystem}. This conflicts with version {required_version} which is the configured version "
372	f"of {subsystem} for this resolve{source}. "
373	"Please remove the `jvm_artifact` target with JVM coordinate "
374	f"{conflicting_coordinate.to_coord_str()}, then re-run "
375	f"`{bin_name()} generate-lockfiles --resolve={resolve_name}`"
376	)
377	super().__init__(msg)	×
378
379
380	class MissingRequiredJvmArtifactsInResolve(ValueError):	9✔
381	def __init__(	9✔
382	self,
383	coordinates: Iterable[Coordinate \| UnversionedCoordinate],
384	*,
385	subsystem: str,
386	resolve_name: str,
387	target_type: str,
388	) -> None:
389	msg = (	×
390	f"The JVM resolve `{resolve_name}` is missing one or more requirements for {subsystem}. "
391	f"Since at least one JVM target type in this repository consumes a `{target_type}` target "
392	"in this resolve, this resolve must contain `jvm_artifact` targets for each requirement of "
393	f"{subsystem}.\n\n"
394	"Please add the following `jvm_artifact` target(s) somewhere in the repository and re-run "
395	f"`{bin_name()} generate-lockfiles --resolve={resolve_name}`:\n"
396	)
397	for coordinate in coordinates:	×
398	if isinstance(coordinate, Coordinate):	×
399	msg += (	×
400	"\njvm_artifact(\n"
401	f' name="{coordinate.group}_{coordinate.artifact}",\n'
402	f' group="{coordinate.group}",\n'
403	f' artifact="{coordinate.artifact}",\n'
404	f' version="{coordinate.version}",\n'
405	f' resolve="{resolve_name}",\n'
406	")\n"
407	)
408	elif isinstance(coordinate, UnversionedCoordinate):	×
409	msg += (	×
410	"\njvm_artifact(\n"
411	f' name="{coordinate.group}_{coordinate.artifact}",\n'
412	f' group="{coordinate.group}",\n'
413	f' artifact="{coordinate.artifact}",\n'
414	' version="<your preferred runtime version>",\n'
415	f' resolve="{resolve_name}",\n'
416	")\n"
417	)
418	super().__init__(msg)	×
419
420
421	def find_jvm_artifacts_or_raise(	9✔
422	required_coordinates: Iterable[Coordinate \| UnversionedCoordinate],
423	resolve: str,
424	jvm_artifact_targets: AllJvmArtifactTargets,
425	jvm: JvmSubsystem,
426	*,
427	subsystem: str,
428	target_type: str,
429	requirement_source: str \| None = None,
430	) -> frozenset[Address]:
431	remaining_coordinates: set[Coordinate \| UnversionedCoordinate] = set(required_coordinates)	7✔
432
433	addresses: set[Address] = set()	7✔
434	for tgt in jvm_artifact_targets:	7✔
435	if tgt[JvmResolveField].normalized_value(jvm) != resolve:	7✔
436	continue	×
437
438	artifact = ArtifactRequirement.from_jvm_artifact_target(tgt)	7✔
439	found_coordinates: set[Coordinate \| UnversionedCoordinate] = set()	7✔
440	for coordinate in remaining_coordinates:	7✔
441	if isinstance(coordinate, Coordinate):	7✔
442	if (	5✔
443	artifact.coordinate.group != coordinate.group
444	or artifact.coordinate.artifact != coordinate.artifact
445	):
446	continue	3✔
447	if artifact.coordinate.version != coordinate.version:	5✔
448	raise ConflictingJvmArtifactVersionInResolveError(	×
449	subsystem=subsystem,
450	requirement_source=requirement_source,
451	resolve_name=resolve,
452	required_version=coordinate.version,
453	conflicting_coordinate=artifact.coordinate,
454	)
455	elif isinstance(coordinate, UnversionedCoordinate):	4✔
456	if (	4✔
457	artifact.coordinate.group != coordinate.group
458	or artifact.coordinate.artifact != coordinate.artifact
459	):
460	continue	3✔
461
462	found_coordinates.add(coordinate)	7✔
463
464	if found_coordinates:	7✔
465	remaining_coordinates.difference_update(found_coordinates)	7✔
466	addresses.add(tgt.address)	7✔
467
468	if remaining_coordinates:	7✔
469	raise MissingRequiredJvmArtifactsInResolve(	×
470	remaining_coordinates,
471	subsystem=subsystem,
472	resolve_name=resolve,
473	target_type=target_type,
474	)
475
476	return frozenset(addresses)	7✔
477
478
479	def rules():	9✔
480	return collect_rules()	9✔

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