19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

54.17

/src/python/pants/util/frozendict.py

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

from __future__ import annotations

from collections.abc import Callable, Iterable, Iterator, Mapping
from typing import Any, TypeVar, cast, overload

from pants.util.memo import memoized_method
from pants.util.strutil import softwrap

K = TypeVar("K")
V = TypeVar("V")


class FrozenDict(Mapping[K, V]):
    """A wrapper around a normal `dict` that removes all methods to mutate the instance and that
    implements __hash__.

    This should be used instead of normal dicts when working with the engine because normal dicts
    are not safe to use.
    """

    @overload
    def __init__(self, __items: Iterable[tuple[K, V]], **kwargs: V) -> None: ...

    @overload
    def __init__(self, __other: Mapping[K, V], **kwargs: V) -> None: ...

    @overload
    def __init__(self, **kwargs: V) -> None: ...

    def __init__(self, *item: Mapping[K, V] | Iterable[tuple[K, V]], **kwargs: V) -> None:
        """Creates a `FrozenDict` with arguments accepted by `dict` that also must be hashable."""
        if len(item) > 1:
            raise ValueError(
                f"{type(self).__name__} was called with {len(item)} positional arguments but it expects one."
            )

        # NB: Keep the variable name `_data` in sync with `externs/mod.rs`.
        self._data = dict(item[0]) if item else dict()
        self._data.update(**kwargs)  # type: ignore[call-overload]

        # NB: We eagerly compute the hash to validate that the values are hashable and to avoid
        # performing the calculation multiple times. This can be revisited if it's found to be a
        # performance bottleneck.
        self._hash = self._calculate_hash()

    @classmethod
    def deep_freeze(cls, data: Mapping[K, V]) -> FrozenDict[K, V]:
        """Convert mutable values to their frozen counter parts.

        Sets and lists are turned into tuples and dicts into FrozenDicts.
        """

        def _freeze(obj):
            if isinstance(obj, dict):
                return cls.deep_freeze(obj)
            if isinstance(obj, (list, set)):
                return tuple(map(_freeze, obj))
            return obj

        return cls({k: _freeze(v) for k, v in data.items()})

    @staticmethod
    def frozen(to_freeze: Mapping[K, V]) -> FrozenDict[K, V]:
        """Returns a `FrozenDict` containing the keys and values of `to_freeze`.

        If `to_freeze` is already a `FrozenDict`, returns the same object.
        """

        return to_freeze if isinstance(to_freeze, FrozenDict) else FrozenDict(to_freeze)

    def __getitem__(self, k: K) -> V:
        return self._data[k]

    def __len__(self) -> int:
        return len(self._data)

    def __iter__(self) -> Iterator[K]:
        return iter(self._data)

    def __reversed__(self) -> Iterator[K]:
        return reversed(tuple(self._data))

    def __eq__(self, other: Any) -> Any:
        # defer to dict's __eq__
        return self._data == other

    def __lt__(self, other: Any) -> bool:
        if not isinstance(other, FrozenDict):
            return NotImplemented
        # If sorting each of these on every __lt__ call ends up being a problem we could consider
        # optimising this, by, for instance, sorting on construction.
        return sorted(self._data.items()) < sorted(other._data.items())

    def __or__(self, other: Any) -> FrozenDict[K, V]:
        if isinstance(other, FrozenDict):
            other = other._data
        elif not isinstance(other, Mapping):
            return NotImplemented
        return FrozenDict(self._data | other)

    def __ror__(self, other: Any) -> FrozenDict[K, V]:
        if isinstance(other, FrozenDict):
            other = other._data
        elif not isinstance(other, Mapping):
            return NotImplemented
        return FrozenDict(other | self._data)

    def _calculate_hash(self) -> int:
        try:
            h = 0
            for pair in self._data.items():
                # xor is commutative, i.e. we get the same hash no matter the order of items. This
                # "relies" on "hash" of the individual elements being unpredictable enough that such
                # a naive aggregation is okay. In addition, the Python hash isn't / shouldn't be
                # used for cryptographically sensitive purposes.
                h ^= hash(pair)
            return h
        except TypeError as e:
            raise TypeError(
                softwrap(
                    f"""
                    Even though you are using a `{type(self).__name__}`, the underlying values are
                    not hashable. Please use hashable (and preferably immutable) types for the
                    underlying values, e.g. use tuples instead of lists and use FrozenOrderedSet
                    instead of set().

                    Original error message: {e}

                    Value: {self}
                    """
                )
            )

    def __hash__(self) -> int:
        return self._hash

    def __repr__(self) -> str:
        return f"{type(self).__name__}({self._data!r})"


class LazyFrozenDict(FrozenDict[K, V]):
    """A lazy version of `FrozenDict` where the values are not loaded until referenced."""

    @overload
    def __init__(
        self, __items: Iterable[tuple[K, Callable[[], V]]], **kwargs: Callable[[], V]
    ) -> None: ...

    @overload
    def __init__(self, __other: Mapping[K, Callable[[], V]], **kwargs: Callable[[], V]) -> None: ...

    @overload
    def __init__(self, **kwargs: Callable[[], V]) -> None: ...

    def __init__(
        self,
        *item: Mapping[K, Callable[[], V]] | Iterable[tuple[K, Callable[[], V]]],
        **kwargs: Callable[[], V],
    ) -> None:
        super().__init__(*item, **kwargs)  # type: ignore[arg-type]

    def __getitem__(self, k: K) -> V:
        return self._get_value(k)

    @memoized_method
    def _get_value(self, k: K) -> V:
        return cast("Callable[[], V]", self._data[k])()

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	1✔
5
6	from collections.abc import Callable, Iterable, Iterator, Mapping	1✔
7	from typing import Any, TypeVar, cast, overload	1✔
8
9	from pants.util.memo import memoized_method	1✔
10	from pants.util.strutil import softwrap	1✔
11
12	K = TypeVar("K")	1✔
13	V = TypeVar("V")	1✔
14
15
16	class FrozenDict(Mapping[K, V]):	1✔
17	"""A wrapper around a normal `dict` that removes all methods to mutate the instance and that
18	implements __hash__.
19
20	This should be used instead of normal dicts when working with the engine because normal dicts
21	are not safe to use.
22	"""
23
24	@overload
25	def __init__(self, __items: Iterable[tuple[K, V]], **kwargs: V) -> None: ...
26
27	@overload
28	def __init__(self, __other: Mapping[K, V], **kwargs: V) -> None: ...
29
30	@overload
31	def __init__(self, **kwargs: V) -> None: ...
32
33	def __init__(self, item: Mapping[K, V] \| Iterable[tuple[K, V]], *kwargs: V) -> None:	1✔
34	"""Creates a `FrozenDict` with arguments accepted by `dict` that also must be hashable."""
35	if len(item) > 1:	1✔
UNCOV 36	raise ValueError(	×
37	f"{type(self).__name__} was called with {len(item)} positional arguments but it expects one."
38	)
39
40	# NB: Keep the variable name `_data` in sync with `externs/mod.rs`.
41	self._data = dict(item[0]) if item else dict()	1✔
42	self._data.update(**kwargs) # type: ignore[call-overload]	1✔
43
44	# NB: We eagerly compute the hash to validate that the values are hashable and to avoid
45	# performing the calculation multiple times. This can be revisited if it's found to be a
46	# performance bottleneck.
47	self._hash = self._calculate_hash()	1✔
48
49	@classmethod	1✔
50	def deep_freeze(cls, data: Mapping[K, V]) -> FrozenDict[K, V]:	1✔
51	"""Convert mutable values to their frozen counter parts.
52
53	Sets and lists are turned into tuples and dicts into FrozenDicts.
54	"""
55
UNCOV 56	def _freeze(obj):	×
UNCOV 57	if isinstance(obj, dict):	×
UNCOV 58	return cls.deep_freeze(obj)	×
UNCOV 59	if isinstance(obj, (list, set)):	×
UNCOV 60	return tuple(map(_freeze, obj))	×
UNCOV 61	return obj	×
62
UNCOV 63	return cls({k: _freeze(v) for k, v in data.items()})	×
64
65	@staticmethod	1✔
66	def frozen(to_freeze: Mapping[K, V]) -> FrozenDict[K, V]:	1✔
67	"""Returns a `FrozenDict` containing the keys and values of `to_freeze`.
68
69	If `to_freeze` is already a `FrozenDict`, returns the same object.
70	"""
71
UNCOV 72	return to_freeze if isinstance(to_freeze, FrozenDict) else FrozenDict(to_freeze)	×
73
74	def __getitem__(self, k: K) -> V:	1✔
UNCOV 75	return self._data[k]	×
76
77	def __len__(self) -> int:	1✔
UNCOV 78	return len(self._data)	×
79
80	def __iter__(self) -> Iterator[K]:	1✔
UNCOV 81	return iter(self._data)	×
82
83	def __reversed__(self) -> Iterator[K]:	1✔
84	return reversed(tuple(self._data))	×
85
86	def __eq__(self, other: Any) -> Any:	1✔
87	# defer to dict's __eq__
UNCOV 88	return self._data == other	×
89
90	def __lt__(self, other: Any) -> bool:	1✔
UNCOV 91	if not isinstance(other, FrozenDict):	×
UNCOV 92	return NotImplemented	×
93	# If sorting each of these on every __lt__ call ends up being a problem we could consider
94	# optimising this, by, for instance, sorting on construction.
UNCOV 95	return sorted(self._data.items()) < sorted(other._data.items())	×
96
97	def __or__(self, other: Any) -> FrozenDict[K, V]:	1✔
UNCOV 98	if isinstance(other, FrozenDict):	×
UNCOV 99	other = other._data	×
UNCOV 100	elif not isinstance(other, Mapping):	×
101	return NotImplemented	×
UNCOV 102	return FrozenDict(self._data \| other)	×
103
104	def __ror__(self, other: Any) -> FrozenDict[K, V]:	1✔
UNCOV 105	if isinstance(other, FrozenDict):	×
106	other = other._data	×
UNCOV 107	elif not isinstance(other, Mapping):	×
108	return NotImplemented	×
UNCOV 109	return FrozenDict(other \| self._data)	×
110
111	def _calculate_hash(self) -> int:	1✔
112	try:	1✔
113	h = 0	1✔
114	for pair in self._data.items():	1✔
115	# xor is commutative, i.e. we get the same hash no matter the order of items. This
116	# "relies" on "hash" of the individual elements being unpredictable enough that such
117	# a naive aggregation is okay. In addition, the Python hash isn't / shouldn't be
118	# used for cryptographically sensitive purposes.
119	h ^= hash(pair)	1✔
120	return h	1✔
UNCOV 121	except TypeError as e:	×
UNCOV 122	raise TypeError(	×
123	softwrap(
124	f"""
125	Even though you are using a `{type(self).__name__}`, the underlying values are
126	not hashable. Please use hashable (and preferably immutable) types for the
127	underlying values, e.g. use tuples instead of lists and use FrozenOrderedSet
128	instead of set().
129
130	Original error message: {e}
131
132	Value: {self}
133	"""
134	)
135	)
136
137	def __hash__(self) -> int:	1✔
UNCOV 138	return self._hash	×
139
140	def __repr__(self) -> str:	1✔
141	return f"{type(self).__name__}({self._data!r})"	1✔
142
143
144	class LazyFrozenDict(FrozenDict[K, V]):	1✔
145	"""A lazy version of `FrozenDict` where the values are not loaded until referenced."""
146
147	@overload
148	def __init__(
149	self, __items: Iterable[tuple[K, Callable[[], V]]], **kwargs: Callable[[], V]
150	) -> None: ...
151
152	@overload
153	def __init__(self, __other: Mapping[K, Callable[[], V]], **kwargs: Callable[[], V]) -> None: ...
154
155	@overload
156	def __init__(self, **kwargs: Callable[[], V]) -> None: ...
157
158	def __init__(	1✔
159	self,
160	*item: Mapping[K, Callable[[], V]] \| Iterable[tuple[K, Callable[[], V]]],
161	**kwargs: Callable[[], V],
162	) -> None:
UNCOV 163	super().__init__(item, *kwargs) # type: ignore[arg-type]	×
164
165	def __getitem__(self, k: K) -> V:	1✔
UNCOV 166	return self._get_value(k)	×
167
168	@memoized_method	1✔
169	def _get_value(self, k: K) -> V:	1✔
UNCOV 170	return cast("Callable[[], V]", self._data[k])()	×

pantsbuild / pants / 19015773527

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