5697667142

Committed 29 Jul 2023 01:09AM UTC coverage: 99.159%. First build

Build # 5697667142

Build Type

push

github

Committed by

chrisrink10

Commit Message

Use Tox 4.0 run command

Run Details

1634 of 1636 branches covered (99.88%)

Branch coverage included in aggregate %.

7686 of 7763 relevant lines covered (99.01%)

0.99 hits per line

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

97.1

/src/basilisp/lang/map.py

from builtins import map as pymap
from typing import Callable, Iterable, Mapping, Optional, Tuple, TypeVar, Union, cast

from immutables import Map as _Map

from basilisp.lang.interfaces import (
    IEvolveableCollection,
    ILispObject,
    IMapEntry,
    IPersistentMap,
    IPersistentVector,
    ISeq,
    ITransientMap,
    IWithMeta,
)
from basilisp.lang.obj import map_lrepr as _map_lrepr
from basilisp.lang.seq import sequence
from basilisp.lang.vector import MapEntry
from basilisp.util import partition

try:
    from immutables._map import MapMutation
except ImportError:
    from immutables.map import MapMutation  # type: ignore[assignment]


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


_ENTRY_SENTINEL = object()


class TransientMap(ITransientMap[K, V]):
    __slots__ = ("_inner",)

    def __init__(self, evolver: "MapMutation[K, V]") -> None:
        self._inner = evolver

    def __bool__(self):
        return True

    def __call__(self, key, default=None):
        return self._inner.get(key, default)

    def __contains__(self, item):
        return item in self._inner

    def __eq__(self, other):
        return self is other

    def __len__(self):
        return len(self._inner)

    def assoc_transient(self, *kvs) -> "TransientMap":
        for k, v in partition(kvs, 2):
            self._inner[k] = v
        return self

    def contains_transient(self, k: K) -> bool:
        return k in self._inner

    def dissoc_transient(self, *ks: K) -> "TransientMap[K, V]":
        for k in ks:
            try:
                del self._inner[k]
            except KeyError:
                pass
        return self

    def entry_transient(self, k: K) -> Optional[IMapEntry[K, V]]:
        v = self._inner.get(k, cast("V", _ENTRY_SENTINEL))
        if v is _ENTRY_SENTINEL:
            return None
        return MapEntry.of(k, v)

    def val_at(self, k, default=None):
        return self._inner.get(k, default)

    def cons_transient(  # type: ignore[override]
        self,
        *elems: Union[
            IPersistentMap[K, V],
            IMapEntry[K, V],
            IPersistentVector[Union[K, V]],
            Mapping[K, V],
        ],
    ) -> "TransientMap[K, V]":
        try:
            for elem in elems:
                if isinstance(elem, (IPersistentMap, Mapping)):
                    for k, v in elem.items():
                        self._inner[k] = v
                elif isinstance(elem, IMapEntry):
                    self._inner[elem.key] = elem.value
                elif elem is None:
                    continue
                else:
                    entry: IMapEntry[K, V] = MapEntry.from_vec(elem)
                    self._inner[entry.key] = entry.value
        except (TypeError, ValueError) as e:
            raise ValueError(
                "Argument to map conj must be another Map or castable to MapEntry"
            ) from e
        else:
            return self

    def to_persistent(self) -> "PersistentMap[K, V]":
        return PersistentMap(self._inner.finish())


class PersistentMap(
    IPersistentMap[K, V], IEvolveableCollection[TransientMap], ILispObject, IWithMeta
):
    """Basilisp Map. Delegates internally to a immutables.Map object.
    Do not instantiate directly. Instead use the m() and map() factory
    methods below."""

    __slots__ = ("_inner", "_meta")

    def __init__(
        self,
        m: "_Map[K, V]",
        meta: Optional[IPersistentMap] = None,
    ) -> None:
        self._inner = m
        self._meta = meta

    @classmethod
    def from_coll(
        cls,
        members: Union[Mapping[K, V], Iterable[Tuple[K, V]]],
        meta: Optional[IPersistentMap] = None,
    ) -> "PersistentMap[K, V]":
        return PersistentMap(_Map(members), meta=meta)

    def __bool__(self):
        return True

    def __call__(self, key, default=None):
        return self._inner.get(key, default)

    def __contains__(self, item):
        return item in self._inner

    def __eq__(self, other):
        if self is other:
            return True
        if not isinstance(other, Mapping):
            return NotImplemented
        if len(self._inner) != len(other):
            return False
        return self._inner == other

    def __getitem__(self, item):
        return self._inner[item]

    def __hash__(self):
        return hash(self._inner)

    def __iter__(self):
        return iter(self._inner)

    def __len__(self):
        return len(self._inner)

    def _lrepr(self, **kwargs):
        return _map_lrepr(
            self._inner.items, start="{", end="}", meta=self._meta, **kwargs
        )

    @property
    def meta(self) -> Optional[IPersistentMap]:
        return self._meta

    def with_meta(self, meta: Optional[IPersistentMap]) -> "PersistentMap":
        return PersistentMap(self._inner, meta=meta)

    def assoc(self, *kvs):
        with self._inner.mutate() as m:
            for k, v in partition(kvs, 2):
                m[k] = v
            return PersistentMap(m.finish())

    def contains(self, k):
        return k in self._inner

    def dissoc(self, *ks):
        with self._inner.mutate() as m:
            for k in ks:
                try:
                    del m[k]
                except KeyError:
                    pass
            return PersistentMap(m.finish())

    def entry(self, k):
        v = self._inner.get(k, cast("V", _ENTRY_SENTINEL))
        if v is _ENTRY_SENTINEL:
            return None
        return MapEntry.of(k, v)

    def val_at(self, k, default=None):
        return self._inner.get(k, default)

    def update(self, *maps: Mapping[K, V]) -> "PersistentMap":
        m: _Map = self._inner.update(*(m.items() for m in maps))
        return PersistentMap(m)

    def update_with(
        self, merge_fn: Callable[[V, V], V], *maps: Mapping[K, V]
    ) -> "PersistentMap[K, V]":
        with self._inner.mutate() as m:
            for map in maps:
                for k, v in map.items():
                    m.set(k, merge_fn(m[k], v) if k in m else v)
            return PersistentMap(m.finish())

    def cons(  # type: ignore[override]
        self,
        *elems: Union[
            IPersistentMap[K, V],
            IMapEntry[K, V],
            IPersistentVector[Union[K, V]],
            Mapping[K, V],
        ],
    ) -> "PersistentMap[K, V]":
        with self._inner.mutate() as m:
            try:
                for elem in elems:
                    if isinstance(elem, (IPersistentMap, Mapping)):
                        for k, v in elem.items():
                            m.set(k, v)
                    elif isinstance(elem, IMapEntry):
                        m.set(elem.key, elem.value)
                    elif elem is None:
                        continue
                    else:
                        entry: IMapEntry[K, V] = MapEntry.from_vec(elem)
                        m.set(entry.key, entry.value)
            except (TypeError, ValueError) as e:
                raise ValueError(
                    "Argument to map conj must be another Map or castable to MapEntry"
                ) from e
            else:
                return PersistentMap(m.finish(), meta=self.meta)

    @staticmethod
    def empty() -> "PersistentMap":
        return EMPTY

    def seq(self) -> Optional[ISeq[IMapEntry[K, V]]]:
        if len(self._inner) == 0:
            return None
        return sequence(MapEntry.of(k, v) for k, v in self._inner.items())

    def to_transient(self) -> TransientMap[K, V]:
        return TransientMap(self._inner.mutate())


EMPTY: PersistentMap = PersistentMap.from_coll(())


def map(  # pylint:disable=redefined-builtin
    kvs: Mapping[K, V], meta: Optional[IPersistentMap] = None
) -> PersistentMap[K, V]:
    """Creates a new map."""
    # For some reason, creating a new `immutables.Map` instance from an existing
    # `basilisp.lang.map.PersistentMap` instance causes issues because the `__iter__`
    # returns only the keys rather than tuple of key/value pairs, even though it
    # adheres to the `Mapping` protocol. Passing the `.items()` directly bypasses
    # this problem.
    return PersistentMap.from_coll(kvs.items(), meta=meta)


def m(**kvs) -> PersistentMap[str, V]:
    """Creates a new map from keyword arguments."""
    return PersistentMap.from_coll(kvs)


def from_entries(entries: Iterable[MapEntry[K, V]]) -> PersistentMap[K, V]:
    with _Map().mutate() as m:  # type: ignore[var-annotated]
        for entry in entries:
            m.set(entry.key, entry.value)
        return PersistentMap(m.finish())


def hash_map(*pairs) -> PersistentMap:
    entries = pymap(lambda v: MapEntry.of(v[0], v[1]), partition(pairs, 2))
    return from_entries(entries)

1	from builtins import map as pymap	1✔
2	from typing import Callable, Iterable, Mapping, Optional, Tuple, TypeVar, Union, cast	1✔
3
4	from immutables import Map as _Map	1✔
5
6	from basilisp.lang.interfaces import (	1✔
7	IEvolveableCollection,
8	ILispObject,
9	IMapEntry,
10	IPersistentMap,
11	IPersistentVector,
12	ISeq,
13	ITransientMap,
14	IWithMeta,
15	)
16	from basilisp.lang.obj import map_lrepr as _map_lrepr	1✔
17	from basilisp.lang.seq import sequence	1✔
18	from basilisp.lang.vector import MapEntry	1✔
19	from basilisp.util import partition	1✔
20
21	try:	1✔
22	from immutables._map import MapMutation	1✔
23	except ImportError:	1✔
24	from immutables.map import MapMutation # type: ignore[assignment]	1✔
25
26
27	T = TypeVar("T")	1✔
28	K = TypeVar("K")	1✔
29	V = TypeVar("V")	1✔
30
31
32	_ENTRY_SENTINEL = object()	1✔
33
34
35	class TransientMap(ITransientMap[K, V]):	1✔
36	__slots__ = ("_inner",)	1✔
37
38	def __init__(self, evolver: "MapMutation[K, V]") -> None:	1✔
39	self._inner = evolver	1✔
40
41	def __bool__(self):	1✔
42	return True	1✔
43
44	def __call__(self, key, default=None):	1✔
45	return self._inner.get(key, default)	1✔
46
47	def __contains__(self, item):	1✔
48	return item in self._inner	1✔
49
50	def __eq__(self, other):	1✔
51	return self is other	1✔
52
53	def __len__(self):	1✔
54	return len(self._inner)	1✔
55
56	def assoc_transient(self, *kvs) -> "TransientMap":	1✔
57	for k, v in partition(kvs, 2):	1✔
58	self._inner[k] = v	1✔
59	return self	1✔
60
61	def contains_transient(self, k: K) -> bool:	1✔
62	return k in self._inner	×
63
64	def dissoc_transient(self, *ks: K) -> "TransientMap[K, V]":	1✔
65	for k in ks:	1✔
66	try:	1✔
67	del self._inner[k]	1✔
68	except KeyError:	1✔
69	pass	1✔
70	return self	1✔
71
72	def entry_transient(self, k: K) -> Optional[IMapEntry[K, V]]:	1✔
73	v = self._inner.get(k, cast("V", _ENTRY_SENTINEL))	×
74	if v is _ENTRY_SENTINEL:	×
75	return None	×
76	return MapEntry.of(k, v)	×
77
78	def val_at(self, k, default=None):	1✔
79	return self._inner.get(k, default)	1✔
80
81	def cons_transient( # type: ignore[override]	1✔
82	self,
83	*elems: Union[
84	IPersistentMap[K, V],
85	IMapEntry[K, V],
86	IPersistentVector[Union[K, V]],
87	Mapping[K, V],
88	],
89	) -> "TransientMap[K, V]":
90	try:	1✔
91	for elem in elems:	1✔
92	if isinstance(elem, (IPersistentMap, Mapping)):	1✔
93	for k, v in elem.items():	1✔
94	self._inner[k] = v	1✔
95	elif isinstance(elem, IMapEntry):	1✔
96	self._inner[elem.key] = elem.value	1✔
97	elif elem is None:	1✔
98	continue	1✔
99	else:
100	entry: IMapEntry[K, V] = MapEntry.from_vec(elem)	1✔
101	self._inner[entry.key] = entry.value	1✔
102	except (TypeError, ValueError) as e:	1✔
103	raise ValueError(	1✔
104	"Argument to map conj must be another Map or castable to MapEntry"
105	) from e
106	else:
107	return self	1✔
108
109	def to_persistent(self) -> "PersistentMap[K, V]":	1✔
110	return PersistentMap(self._inner.finish())	1✔
111
112
113	class PersistentMap(	1✔
114	IPersistentMap[K, V], IEvolveableCollection[TransientMap], ILispObject, IWithMeta
115	):
116	"""Basilisp Map. Delegates internally to a immutables.Map object.
117	Do not instantiate directly. Instead use the m() and map() factory
118	methods below."""
119
120	__slots__ = ("_inner", "_meta")	1✔
121
122	def __init__(	1✔
123	self,
124	m: "_Map[K, V]",
125	meta: Optional[IPersistentMap] = None,
126	) -> None:
127	self._inner = m	1✔
128	self._meta = meta	1✔
129
130	@classmethod	1✔
131	def from_coll(	1✔
132	cls,
133	members: Union[Mapping[K, V], Iterable[Tuple[K, V]]],
134	meta: Optional[IPersistentMap] = None,
135	) -> "PersistentMap[K, V]":
136	return PersistentMap(_Map(members), meta=meta)	1✔
137
138	def __bool__(self):	1✔
139	return True	1✔
140
141	def __call__(self, key, default=None):	1✔
142	return self._inner.get(key, default)	1✔
143
144	def __contains__(self, item):	1✔
145	return item in self._inner	1✔
146
147	def __eq__(self, other):	1✔
148	if self is other:	1✔
149	return True	1✔
150	if not isinstance(other, Mapping):	1✔
151	return NotImplemented	1✔
152	if len(self._inner) != len(other):	1✔
153	return False	1✔
154	return self._inner == other	1✔
155
156	def __getitem__(self, item):	1✔
157	return self._inner[item]	1✔
158
159	def __hash__(self):	1✔
160	return hash(self._inner)	1✔
161
162	def __iter__(self):	1✔
163	return iter(self._inner)	1✔
164
165	def __len__(self):	1✔
166	return len(self._inner)	1✔
167
168	def _lrepr(self, **kwargs):	1✔
169	return _map_lrepr(	1✔
170	self._inner.items, start="{", end="}", meta=self._meta, **kwargs
171	)
172
173	@property	1✔
174	def meta(self) -> Optional[IPersistentMap]:	1✔
175	return self._meta	1✔
176
177	def with_meta(self, meta: Optional[IPersistentMap]) -> "PersistentMap":	1✔
178	return PersistentMap(self._inner, meta=meta)	1✔
179
180	def assoc(self, *kvs):	1✔
181	with self._inner.mutate() as m:	1✔
182	for k, v in partition(kvs, 2):	1✔
183	m[k] = v	1✔
184	return PersistentMap(m.finish())	1✔
185
186	def contains(self, k):	1✔
187	return k in self._inner	1✔
188
189	def dissoc(self, *ks):	1✔
190	with self._inner.mutate() as m:	1✔
191	for k in ks:	1✔
192	try:	1✔
193	del m[k]	1✔
194	except KeyError:	1✔
195	pass	1✔
196	return PersistentMap(m.finish())	1✔
197
198	def entry(self, k):	1✔
199	v = self._inner.get(k, cast("V", _ENTRY_SENTINEL))	1✔
200	if v is _ENTRY_SENTINEL:	1✔
201	return None	1✔
202	return MapEntry.of(k, v)	1✔
203
204	def val_at(self, k, default=None):	1✔
205	return self._inner.get(k, default)	1✔
206
207	def update(self, *maps: Mapping[K, V]) -> "PersistentMap":	1✔
208	m: _Map = self._inner.update(*(m.items() for m in maps))	1✔
209	return PersistentMap(m)	1✔
210
211	def update_with(	1✔
212	self, merge_fn: Callable[[V, V], V], *maps: Mapping[K, V]
213	) -> "PersistentMap[K, V]":
214	with self._inner.mutate() as m:	1✔
215	for map in maps:	1✔
216	for k, v in map.items():	1✔
217	m.set(k, merge_fn(m[k], v) if k in m else v)	1✔
218	return PersistentMap(m.finish())	1✔
219
220	def cons( # type: ignore[override]	1✔
221	self,
222	*elems: Union[
223	IPersistentMap[K, V],
224	IMapEntry[K, V],
225	IPersistentVector[Union[K, V]],
226	Mapping[K, V],
227	],
228	) -> "PersistentMap[K, V]":
229	with self._inner.mutate() as m:	1✔
230	try:	1✔
231	for elem in elems:	1✔
232	if isinstance(elem, (IPersistentMap, Mapping)):	1✔
233	for k, v in elem.items():	1✔
234	m.set(k, v)	1✔
235	elif isinstance(elem, IMapEntry):	1✔
236	m.set(elem.key, elem.value)	1✔
237	elif elem is None:	1✔
238	continue	1✔
239	else:
240	entry: IMapEntry[K, V] = MapEntry.from_vec(elem)	1✔
241	m.set(entry.key, entry.value)	1✔
242	except (TypeError, ValueError) as e:	1✔
243	raise ValueError(	1✔
244	"Argument to map conj must be another Map or castable to MapEntry"
245	) from e
246	else:
247	return PersistentMap(m.finish(), meta=self.meta)	1✔
248
249	@staticmethod	1✔
250	def empty() -> "PersistentMap":	1✔
251	return EMPTY	1✔
252
253	def seq(self) -> Optional[ISeq[IMapEntry[K, V]]]:	1✔
254	if len(self._inner) == 0:	1✔
255	return None	1✔
256	return sequence(MapEntry.of(k, v) for k, v in self._inner.items())	1✔
257
258	def to_transient(self) -> TransientMap[K, V]:	1✔
259	return TransientMap(self._inner.mutate())	1✔
260
261
262	EMPTY: PersistentMap = PersistentMap.from_coll(())	1✔
263
264
265	def map( # pylint:disable=redefined-builtin	1✔
266	kvs: Mapping[K, V], meta: Optional[IPersistentMap] = None
267	) -> PersistentMap[K, V]:
268	"""Creates a new map."""
269	# For some reason, creating a new `immutables.Map` instance from an existing
270	# `basilisp.lang.map.PersistentMap` instance causes issues because the `__iter__`
271	# returns only the keys rather than tuple of key/value pairs, even though it
272	# adheres to the `Mapping` protocol. Passing the `.items()` directly bypasses
273	# this problem.
274	return PersistentMap.from_coll(kvs.items(), meta=meta)	1✔
275
276
277	def m(**kvs) -> PersistentMap[str, V]:	1✔
278	"""Creates a new map from keyword arguments."""
279	return PersistentMap.from_coll(kvs)	1✔
280
281
282	def from_entries(entries: Iterable[MapEntry[K, V]]) -> PersistentMap[K, V]:	1✔
283	with _Map().mutate() as m: # type: ignore[var-annotated]	1✔
284	for entry in entries:	1✔
285	m.set(entry.key, entry.value)	1✔
286	return PersistentMap(m.finish())	1✔
287
288
289	def hash_map(*pairs) -> PersistentMap:	1✔
290	entries = pymap(lambda v: MapEntry.of(v[0], v[1]), partition(pairs, 2))	1✔
291	return from_entries(entries)	1✔

basilisp-lang / basilisp / 5697667142

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