18517631058

Committed 15 Oct 2025 04:18AM UTC coverage: 69.207% (-11.1%) from 80.267%

Build # 18517631058

Build Type

Pull #22745

github

Committed by

web-flow

Commit Message

Merge 642a76ca1 into 99919310e

Pull Request Pull Request #22745: [windows] Add windows support in the stdio crate.

Run Details

53815 of 77759 relevant lines covered (69.21%)

2.42 hits per line

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

73.61

/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)

        # 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	7✔
5
6	from collections.abc import Callable, Iterable, Iterator, Mapping	7✔
7	from typing import Any, TypeVar, cast, overload	7✔
8
9	from pants.util.memo import memoized_method	7✔
10	from pants.util.strutil import softwrap	7✔
11
12	K = TypeVar("K")	7✔
13	V = TypeVar("V")	7✔
14
15
16	class FrozenDict(Mapping[K, V]):	7✔
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:	7✔
34	"""Creates a `FrozenDict` with arguments accepted by `dict` that also must be hashable."""
35	if len(item) > 1:	7✔
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()	7✔
42	self._data.update(**kwargs)	7✔
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()	7✔
48
49	@classmethod	7✔
50	def deep_freeze(cls, data: Mapping[K, V]) -> FrozenDict[K, V]:	7✔
51	"""Convert mutable values to their frozen counter parts.
52
53	Sets and lists are turned into tuples and dicts into FrozenDicts.
54	"""
55
56	def _freeze(obj):	3✔
57	if isinstance(obj, dict):	2✔
58	return cls.deep_freeze(obj)	×
59	if isinstance(obj, (list, set)):	2✔
60	return tuple(map(_freeze, obj))	1✔
61	return obj	2✔
62
63	return cls({k: _freeze(v) for k, v in data.items()})	3✔
64
65	@staticmethod	7✔
66	def frozen(to_freeze: Mapping[K, V]) -> FrozenDict[K, V]:	7✔
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
72	return to_freeze if isinstance(to_freeze, FrozenDict) else FrozenDict(to_freeze)	×
73
74	def __getitem__(self, k: K) -> V:	7✔
75	return self._data[k]	7✔
76
77	def __len__(self) -> int:	7✔
78	return len(self._data)	7✔
79
80	def __iter__(self) -> Iterator[K]:	7✔
81	return iter(self._data)	7✔
82
83	def __reversed__(self) -> Iterator[K]:	7✔
84	return reversed(tuple(self._data))	×
85
86	def __eq__(self, other: Any) -> Any:	7✔
87	# defer to dict's __eq__
88	return self._data == other	7✔
89
90	def __lt__(self, other: Any) -> bool:	7✔
91	if not isinstance(other, FrozenDict):	×
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.
95	return sorted(self._data.items()) < sorted(other._data.items())	×
96
97	def __or__(self, other: Any) -> FrozenDict[K, V]:	7✔
98	if isinstance(other, FrozenDict):	×
99	other = other._data	×
100	elif not isinstance(other, Mapping):	×
101	return NotImplemented	×
102	return FrozenDict(self._data \| other)	×
103
104	def __ror__(self, other: Any) -> FrozenDict[K, V]:	7✔
105	if isinstance(other, FrozenDict):	×
106	other = other._data	×
107	elif not isinstance(other, Mapping):	×
108	return NotImplemented	×
109	return FrozenDict(other \| self._data)	×
110
111	def _calculate_hash(self) -> int:	7✔
112	try:	7✔
113	h = 0	7✔
114	for pair in self._data.items():	7✔
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)	7✔
120	return h	7✔
121	except TypeError as e:	×
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:	7✔
138	return self._hash	7✔
139
140	def __repr__(self) -> str:	7✔
141	return f"{type(self).__name__}({self._data!r})"	7✔
142
143
144	class LazyFrozenDict(FrozenDict[K, V]):	7✔
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__(	7✔
159	self,
160	*item: Mapping[K, Callable[[], V]] \| Iterable[tuple[K, Callable[[], V]]],
161	**kwargs: Callable[[], V],
162	) -> None:
163	super().__init__(item, *kwargs) # type: ignore[arg-type]	2✔
164
165	def __getitem__(self, k: K) -> V:	7✔
166	return self._get_value(k)	2✔
167
168	@memoized_method	7✔
169	def _get_value(self, k: K) -> V:	7✔
170	return cast("Callable[[], V]", self._data[k])()	2✔

pantsbuild / pants / 18517631058

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