12281612501

Committed 11 Dec 2024 05:24PM CUT coverage: 98.677% (-0.06%) from 98.738%

Build # 12281612501

Build Type

Pull #1172

github

Committed by

web-flow

Commit Message

Merge a45766c9a into 2dffcc004

Pull Request Pull Request #1172: Derive module (test) names from `ns` declarations

Run Details

1026 of 1035 branches covered (99.13%)

Branch coverage included in aggregate %.

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

5 existing lines in 1 file now uncovered.

8820 of 8943 relevant lines covered (98.62%)

0.99 hits per line

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

96.74

/src/basilisp/lang/vector.py

from collections.abc import Iterable, Sequence
from functools import total_ordering
from typing import Optional, TypeVar, Union, cast, overload

from pyrsistent import PVector, pvector  # noqa # pylint: disable=unused-import
from pyrsistent.typing import PVectorEvolver
from typing_extensions import Unpack

from basilisp.lang.interfaces import (
    IEvolveableCollection,
    ILispObject,
    IMapEntry,
    IPersistentMap,
    IPersistentVector,
    IReduce,
    IReduceKV,
    ISeq,
    ITransientVector,
    IWithMeta,
    ReduceFunction,
    ReduceKVFunction,
    seq_equals,
)
from basilisp.lang.obj import PrintSettings
from basilisp.lang.obj import seq_lrepr as _seq_lrepr
from basilisp.lang.reduced import Reduced
from basilisp.lang.seq import sequence
from basilisp.util import partition

T = TypeVar("T")

T_reduce = TypeVar("T_reduce")
V_contra = TypeVar("V_contra", contravariant=True)


class TransientVector(ITransientVector[T]):
    __slots__ = ("_inner",)

    def __init__(self, wrapped: "PVectorEvolver[T]") -> None:
        self._inner = wrapped

    def __bool__(self):
        return True

    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 cons_transient(self, *elems: T) -> "TransientVector[T]":  # type: ignore[override]
        for elem in elems:
            self._inner.append(elem)
        return self

    def assoc_transient(self, *kvs: T) -> "TransientVector[T]":
        for i, v in cast("Sequence[tuple[int, T]]", partition(kvs, 2)):
            self._inner.set(i, v)
        return self

    def contains_transient(self, k: int) -> bool:
        return 0 <= k < len(self._inner)

    def entry_transient(self, k: int) -> Optional[IMapEntry[int, T]]:
        try:
            return MapEntry.of(k, self._inner[k])
        except IndexError:
            return None

    def val_at(self, k: int, default=None):
        try:
            return self._inner[k]
        except IndexError:
            return default

    def pop_transient(self) -> "TransientVector[T]":
        if len(self) == 0:
            raise IndexError("Cannot pop an empty vector")
        del self._inner[-1]
        return self

    def to_persistent(self) -> "PersistentVector[T]":
        return PersistentVector(self._inner.persistent())


@total_ordering
class PersistentVector(
    IReduce,
    IReduceKV,
    IPersistentVector[T],
    IEvolveableCollection[TransientVector],
    ILispObject,
    IWithMeta,
):
    """Basilisp Vector. Delegates internally to a pyrsistent.PVector object.
    Do not instantiate directly. Instead use the v() and vec() factory
    methods below."""

    __slots__ = ("_inner", "_meta")

    def __init__(
        self, wrapped: "PVector[T]", meta: Optional[IPersistentMap] = None
    ) -> None:
        self._inner = wrapped
        self._meta = meta

    def __bool__(self):
        return True

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

    def __eq__(self, other):
        if self is other:
            return True
        if hasattr(other, "__len__") and len(self) != len(other):
            return False
        return seq_equals(self, other)

    def __getitem__(self, item):
        if isinstance(item, slice):
            return PersistentVector(self._inner[item])
        return self._inner[item]

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

    def __iter__(self):
        yield from self._inner

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

    def __call__(self, k: int, default: Optional[T] = None) -> Optional[T]:
        return self.val_at(k, default=default)

    def __lt__(self, other):
        """Return true if the `self` vector is shorter than the
        `other` vector, or the first unequal element in `self` when
        iterating from left to right is less than the corresponding
        `other` element.

        This is to support the comparing and sorting operations of
        vectors in Clojure."""

        if other is None:  # pragma: no cover
            return False
        if not isinstance(other, PersistentVector):
            return NotImplemented
        if len(self) != len(other):
            return len(self) < len(other)

        for x, y in zip(self, other):
            if x < y:
                return True
            elif y < x:
                return False
        return False

    def _lrepr(self, **kwargs: Unpack[PrintSettings]) -> str:
        return _seq_lrepr(self._inner, "[", "]", meta=self._meta, **kwargs)

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

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

    def cons(self, *elems: T) -> "PersistentVector[T]":  # type: ignore[override]
        e = self._inner.evolver()
        for elem in elems:
            e.append(elem)
        return PersistentVector(e.persistent(), meta=self.meta)

    def assoc(self, *kvs: T) -> "PersistentVector[T]":
        return PersistentVector(self._inner.mset(*kvs), meta=self._meta)  # type: ignore[arg-type]

    def contains(self, k: int) -> bool:
        return 0 <= k < len(self._inner)

    def entry(self, k: int) -> Optional[IMapEntry[int, T]]:
        try:
            return MapEntry.of(k, self._inner[k])
        except IndexError:
            return None

    def val_at(self, k: int, default: Optional[T] = None) -> Optional[T]:
        try:
            return self._inner[k]
        except (IndexError, TypeError):
            return default

    def empty(self) -> "PersistentVector[T]":
        return EMPTY.with_meta(self._meta)

    def seq(self) -> Optional[ISeq[T]]:  # type: ignore[override]
        if len(self._inner) == 0:
            return None
        return sequence(self)

    def peek(self) -> Optional[T]:
        if len(self) == 0:
            return None
        return self[-1]

    def pop(self) -> "PersistentVector[T]":
        if len(self) == 0:
            raise IndexError("Cannot pop an empty vector")
        return self[:-1]

    def rseq(self) -> ISeq[T]:
        return sequence(reversed(self))

    def to_transient(self) -> TransientVector:
        return TransientVector(self._inner.evolver())

    @overload
    def reduce(self, f: ReduceFunction[T_reduce, V_contra]) -> T_reduce: ...

    @overload
    def reduce(  # pylint: disable=arguments-differ
        self, f: ReduceFunction[T_reduce, V_contra], init: T_reduce
    ) -> T_reduce: ...

    def reduce(self, f, init=IReduce.REDUCE_SENTINEL):
        if init is IReduce.REDUCE_SENTINEL:
            if len(self) == 0:
                return f()
            else:
                init = self._inner[0]
                for item in self._inner[1:]:
                    init = f(init, item)
                    if isinstance(init, Reduced):
                        return init.deref()
        else:
            for item in self._inner:
                init = f(init, item)
                if isinstance(init, Reduced):
                    return init.deref()
        return init

    def reduce_kv(self, f: ReduceKVFunction, init: T_reduce) -> T_reduce:
        for idx, item in enumerate(self._inner):
            init = f(init, idx, item)
            if isinstance(init, Reduced):
                return init.deref()
        return init


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


class MapEntry(IMapEntry[K, V], PersistentVector[Union[K, V]]):
    __slots__ = ()

    def __init__(self, wrapped: "PVector[Union[K, V]]") -> None:
        try:
            if not len(wrapped) == 2:
                raise ValueError("Vector arg to map conj must be a pair")
        except TypeError as e:
            raise TypeError(f"Cannot make map entry from {type(wrapped)}") from e

        super().__init__(wrapped)

    @property
    def key(self) -> K:
        return self[0]

    @property
    def value(self) -> V:
        return self[1]

    @staticmethod
    def of(k: K, v: V) -> "MapEntry[K, V]":
        return MapEntry(pvector([k, v]))

    @staticmethod
    def from_vec(v: Sequence[Union[K, V]]) -> "MapEntry[K, V]":
        return MapEntry(pvector(v))


EMPTY: PersistentVector = PersistentVector(pvector(()))


def vector(
    members: Iterable[T], meta: Optional[IPersistentMap] = None
) -> PersistentVector[T]:
    """Creates a new vector."""
    return PersistentVector(pvector(members), meta=meta)


def v(*members: T, meta: Optional[IPersistentMap] = None) -> PersistentVector[T]:
    """Creates a new vector from members."""
    return PersistentVector(pvector(members), meta=meta)

1	from collections.abc import Iterable, Sequence	1✔
2	from functools import total_ordering	1✔
3	from typing import Optional, TypeVar, Union, cast, overload	1✔
4
5	from pyrsistent import PVector, pvector # noqa # pylint: disable=unused-import	1✔
6	from pyrsistent.typing import PVectorEvolver	1✔
7	from typing_extensions import Unpack	1✔
8
9	from basilisp.lang.interfaces import (	1✔
10	IEvolveableCollection,
11	ILispObject,
12	IMapEntry,
13	IPersistentMap,
14	IPersistentVector,
15	IReduce,
16	IReduceKV,
17	ISeq,
18	ITransientVector,
19	IWithMeta,
20	ReduceFunction,
21	ReduceKVFunction,
22	seq_equals,
23	)
24	from basilisp.lang.obj import PrintSettings	1✔
25	from basilisp.lang.obj import seq_lrepr as _seq_lrepr	1✔
26	from basilisp.lang.reduced import Reduced	1✔
27	from basilisp.lang.seq import sequence	1✔
28	from basilisp.util import partition	1✔
29
30	T = TypeVar("T")	1✔
31
32	T_reduce = TypeVar("T_reduce")	1✔
33	V_contra = TypeVar("V_contra", contravariant=True)	1✔
34
35
36	class TransientVector(ITransientVector[T]):	1✔
37	__slots__ = ("_inner",)	1✔
38
39	def __init__(self, wrapped: "PVectorEvolver[T]") -> None:	1✔
40	self._inner = wrapped	1✔
41
42	def __bool__(self):	1✔
43	return True	×
44
45	def __contains__(self, item):	1✔
46	return item in self._inner	×
47
48	def __eq__(self, other):	1✔
49	return self is other	1✔
50
51	def __len__(self):	1✔
52	return len(self._inner)	1✔
53
54	def cons_transient(self, *elems: T) -> "TransientVector[T]": # type: ignore[override]	1✔
55	for elem in elems:	1✔
56	self._inner.append(elem)	1✔
57	return self	1✔
58
59	def assoc_transient(self, *kvs: T) -> "TransientVector[T]":	1✔
60	for i, v in cast("Sequence[tuple[int, T]]", partition(kvs, 2)):	1✔
61	self._inner.set(i, v)	1✔
62	return self	1✔
63
64	def contains_transient(self, k: int) -> bool:	1✔
65	return 0 <= k < len(self._inner)	1✔
66
67	def entry_transient(self, k: int) -> Optional[IMapEntry[int, T]]:	1✔
68	try:	×
69	return MapEntry.of(k, self._inner[k])	×
70	except IndexError:	×
71	return None	×
72
73	def val_at(self, k: int, default=None):	1✔
74	try:	1✔
75	return self._inner[k]	1✔
76	except IndexError:	1✔
77	return default	1✔
78
79	def pop_transient(self) -> "TransientVector[T]":	1✔
80	if len(self) == 0:	1✔
81	raise IndexError("Cannot pop an empty vector")	1✔
82	del self._inner[-1]	1✔
83	return self	1✔
84
85	def to_persistent(self) -> "PersistentVector[T]":	1✔
86	return PersistentVector(self._inner.persistent())	1✔
87
88
89	@total_ordering	1✔
90	class PersistentVector(	1✔
91	IReduce,
92	IReduceKV,
93	IPersistentVector[T],
94	IEvolveableCollection[TransientVector],
95	ILispObject,
96	IWithMeta,
97	):
98	"""Basilisp Vector. Delegates internally to a pyrsistent.PVector object.
99	Do not instantiate directly. Instead use the v() and vec() factory
100	methods below."""
101
102	__slots__ = ("_inner", "_meta")	1✔
103
104	def __init__(	1✔
105	self, wrapped: "PVector[T]", meta: Optional[IPersistentMap] = None
106	) -> None:
107	self._inner = wrapped	1✔
108	self._meta = meta	1✔
109
110	def __bool__(self):	1✔
111	return True	1✔
112
113	def __contains__(self, item):	1✔
114	return item in self._inner	1✔
115
116	def __eq__(self, other):	1✔
117	if self is other:	1✔
118	return True	1✔
119	if hasattr(other, "__len__") and len(self) != len(other):	1✔
120	return False	1✔
121	return seq_equals(self, other)	1✔
122
123	def __getitem__(self, item):	1✔
124	if isinstance(item, slice):	1✔
125	return PersistentVector(self._inner[item])	1✔
126	return self._inner[item]	1✔
127
128	def __hash__(self):	1✔
129	return hash(self._inner)	1✔
130
131	def __iter__(self):	1✔
132	yield from self._inner	1✔
133
134	def __len__(self):	1✔
135	return len(self._inner)	1✔
136
137	def __call__(self, k: int, default: Optional[T] = None) -> Optional[T]:	1✔
138	return self.val_at(k, default=default)	1✔
139
140	def __lt__(self, other):	1✔
141	"""Return true if the `self` vector is shorter than the
142	`other` vector, or the first unequal element in `self` when
143	iterating from left to right is less than the corresponding
144	`other` element.
145
146	This is to support the comparing and sorting operations of
147	vectors in Clojure."""
148
149	if other is None: # pragma: no cover
150	return False
151	if not isinstance(other, PersistentVector):	1✔
152	return NotImplemented	1✔
153	if len(self) != len(other):	1✔
154	return len(self) < len(other)	1✔
155
156	for x, y in zip(self, other):	1✔
157	if x < y:	1✔
158	return True	1✔
159	elif y < x:	1✔
160	return False	1✔
161	return False	1✔
162
163	def _lrepr(self, **kwargs: Unpack[PrintSettings]) -> str:	1✔
164	return _seq_lrepr(self._inner, "[", "]", meta=self._meta, **kwargs)	1✔
165
166	@property	1✔
167	def meta(self) -> Optional[IPersistentMap]:	1✔
168	return self._meta	1✔
169
170	def with_meta(self, meta: Optional[IPersistentMap]) -> "PersistentVector[T]":	1✔
171	return vector(self._inner, meta=meta)	1✔
172
173	def cons(self, *elems: T) -> "PersistentVector[T]": # type: ignore[override]	1✔
174	e = self._inner.evolver()	1✔
175	for elem in elems:	1✔
176	e.append(elem)	1✔
177	return PersistentVector(e.persistent(), meta=self.meta)	1✔
178
179	def assoc(self, *kvs: T) -> "PersistentVector[T]":	1✔
180	return PersistentVector(self._inner.mset(*kvs), meta=self._meta) # type: ignore[arg-type]	1✔
181
182	def contains(self, k: int) -> bool:	1✔
183	return 0 <= k < len(self._inner)	1✔
184
185	def entry(self, k: int) -> Optional[IMapEntry[int, T]]:	1✔
186	try:	1✔
187	return MapEntry.of(k, self._inner[k])	1✔
188	except IndexError:	1✔
189	return None	1✔
190
191	def val_at(self, k: int, default: Optional[T] = None) -> Optional[T]:	1✔
192	try:	1✔
193	return self._inner[k]	1✔
194	except (IndexError, TypeError):	1✔
195	return default	1✔
196
197	def empty(self) -> "PersistentVector[T]":	1✔
198	return EMPTY.with_meta(self._meta)	1✔
199
200	def seq(self) -> Optional[ISeq[T]]: # type: ignore[override]	1✔
201	if len(self._inner) == 0:	1✔
202	return None	1✔
203	return sequence(self)	1✔
204
205	def peek(self) -> Optional[T]:	1✔
206	if len(self) == 0:	1✔
207	return None	1✔
208	return self[-1]	1✔
209
210	def pop(self) -> "PersistentVector[T]":	1✔
211	if len(self) == 0:	1✔
212	raise IndexError("Cannot pop an empty vector")	1✔
213	return self[:-1]	1✔
214
215	def rseq(self) -> ISeq[T]:	1✔
216	return sequence(reversed(self))	1✔
217
218	def to_transient(self) -> TransientVector:	1✔
219	return TransientVector(self._inner.evolver())	1✔
220
221	@overload	1✔
222	def reduce(self, f: ReduceFunction[T_reduce, V_contra]) -> T_reduce: ...	1✔
223
224	@overload	1✔
225	def reduce( # pylint: disable=arguments-differ	1✔
226	self, f: ReduceFunction[T_reduce, V_contra], init: T_reduce
227	) -> T_reduce: ...
228
229	def reduce(self, f, init=IReduce.REDUCE_SENTINEL):	1✔
230	if init is IReduce.REDUCE_SENTINEL:	1✔
231	if len(self) == 0:	1✔
232	return f()	1✔
233	else:
234	init = self._inner[0]	1✔
235	for item in self._inner[1:]:	1✔
236	init = f(init, item)	1✔
237	if isinstance(init, Reduced):	1✔
238	return init.deref()	1✔
239	else:
240	for item in self._inner:	1✔
241	init = f(init, item)	1✔
242	if isinstance(init, Reduced):	1✔
243	return init.deref()	1✔
244	return init	1✔
245
246	def reduce_kv(self, f: ReduceKVFunction, init: T_reduce) -> T_reduce:	1✔
247	for idx, item in enumerate(self._inner):	1✔
248	init = f(init, idx, item)	1✔
249	if isinstance(init, Reduced):	1✔
250	return init.deref()	1✔
251	return init	1✔
252
253
254	K = TypeVar("K")	1✔
255	V = TypeVar("V")	1✔
256
257
258	class MapEntry(IMapEntry[K, V], PersistentVector[Union[K, V]]):	1✔
259	__slots__ = ()	1✔
260
261	def __init__(self, wrapped: "PVector[Union[K, V]]") -> None:	1✔
262	try:	1✔
263	if not len(wrapped) == 2:	1✔
264	raise ValueError("Vector arg to map conj must be a pair")	1✔
265	except TypeError as e:	1✔
266	raise TypeError(f"Cannot make map entry from {type(wrapped)}") from e	×
267
268	super().__init__(wrapped)	1✔
269
270	@property	1✔
271	def key(self) -> K:	1✔
272	return self[0]	1✔
273
274	@property	1✔
275	def value(self) -> V:	1✔
276	return self[1]	1✔
277
278	@staticmethod	1✔
279	def of(k: K, v: V) -> "MapEntry[K, V]":	1✔
280	return MapEntry(pvector([k, v]))	1✔
281
282	@staticmethod	1✔
283	def from_vec(v: Sequence[Union[K, V]]) -> "MapEntry[K, V]":	1✔
284	return MapEntry(pvector(v))	1✔
285
286
287	EMPTY: PersistentVector = PersistentVector(pvector(()))	1✔
288
289
290	def vector(	1✔
291	members: Iterable[T], meta: Optional[IPersistentMap] = None
292	) -> PersistentVector[T]:
293	"""Creates a new vector."""
294	return PersistentVector(pvector(members), meta=meta)	1✔
295
296
297	def v(*members: T, meta: Optional[IPersistentMap] = None) -> PersistentVector[T]:	1✔
298	"""Creates a new vector from members."""
299	return PersistentVector(pvector(members), meta=meta)	1✔

basilisp-lang / basilisp / 12281612501

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