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

95.98

/src/basilisp/lang/vector.py

from functools import total_ordering
from typing import Iterable, Optional, Sequence, TypeVar, Union

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

from basilisp.lang.interfaces import (
    IEvolveableCollection,
    ILispObject,
    IMapEntry,
    IPersistentMap,
    IPersistentVector,
    ISeq,
    ITransientVector,
    IWithMeta,
    seq_equals,
)
from basilisp.lang.obj import seq_lrepr as _seq_lrepr
from basilisp.lang.seq import sequence
from basilisp.util import partition

T = TypeVar("T")


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 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(
    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 __lt__(self, other):
        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)
        return any(x < y for x, y in zip(self, other))

    def _lrepr(self, **kwargs) -> 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))  # type: ignore[arg-type]

    def contains(self, k):
        return 0 <= k < len(self._inner)

    def entry(self, k):
        try:
            return MapEntry.of(k, self._inner[k])
        except IndexError:
            return None

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

    @staticmethod
    def empty() -> "PersistentVector[T]":
        return EMPTY

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


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 functools import total_ordering	1✔
2	from typing import Iterable, Optional, Sequence, TypeVar, Union	1✔
3
4	from pyrsistent import PVector, pvector # noqa # pylint: disable=unused-import	1✔
5	from pyrsistent.typing import PVectorEvolver	1✔
6
7	from basilisp.lang.interfaces import (	1✔
8	IEvolveableCollection,
9	ILispObject,
10	IMapEntry,
11	IPersistentMap,
12	IPersistentVector,
13	ISeq,
14	ITransientVector,
15	IWithMeta,
16	seq_equals,
17	)
18	from basilisp.lang.obj import seq_lrepr as _seq_lrepr	1✔
19	from basilisp.lang.seq import sequence	1✔
20	from basilisp.util import partition	1✔
21
22	T = TypeVar("T")	1✔
23
24
25	class TransientVector(ITransientVector[T]):	1✔
26	__slots__ = ("_inner",)	1✔
27
28	def __init__(self, wrapped: "PVectorEvolver[T]") -> None:	1✔
29	self._inner = wrapped	1✔
30
31	def __bool__(self):	1✔
32	return True	1✔
33
34	def __contains__(self, item):	1✔
35	return item in self._inner	×
36
37	def __eq__(self, other):	1✔
38	return self is other	1✔
39
40	def __len__(self):	1✔
41	return len(self._inner)	1✔
42
43	def cons_transient(self, *elems: T) -> "TransientVector[T]": # type: ignore[override]	1✔
44	for elem in elems:	1✔
45	self._inner.append(elem)	1✔
46	return self	1✔
47
48	def assoc_transient(self, *kvs: T) -> "TransientVector[T]":	1✔
49	for i, v in partition(kvs, 2):	1✔
50	self._inner.set(i, v)	1✔
51	return self	1✔
52
53	def contains_transient(self, k: int) -> bool:	1✔
54	return 0 <= k < len(self._inner)	×
55
56	def entry_transient(self, k: int) -> Optional[IMapEntry[int, T]]:	1✔
57	try:	×
58	return MapEntry.of(k, self._inner[k])	×
59	except IndexError:	×
60	return None	×
61
62	def val_at(self, k: int, default=None):	1✔
63	try:	1✔
64	return self._inner[k]	1✔
65	except IndexError:	1✔
66	return default	1✔
67
68	def pop_transient(self) -> "TransientVector[T]":	1✔
69	if len(self) == 0:	1✔
70	raise IndexError("Cannot pop an empty vector")	1✔
71	del self._inner[-1]	1✔
72	return self	1✔
73
74	def to_persistent(self) -> "PersistentVector[T]":	1✔
75	return PersistentVector(self._inner.persistent())	1✔
76
77
78	@total_ordering	1✔
79	class PersistentVector(	1✔
80	IPersistentVector[T], IEvolveableCollection[TransientVector], ILispObject, IWithMeta
81	):
82	"""Basilisp Vector. Delegates internally to a pyrsistent.PVector object.
83	Do not instantiate directly. Instead use the v() and vec() factory
84	methods below."""
85
86	__slots__ = ("_inner", "_meta")	1✔
87
88	def __init__(	1✔
89	self, wrapped: "PVector[T]", meta: Optional[IPersistentMap] = None
90	) -> None:
91	self._inner = wrapped	1✔
92	self._meta = meta	1✔
93
94	def __bool__(self):	1✔
95	return True	1✔
96
97	def __contains__(self, item):	1✔
98	return item in self._inner	1✔
99
100	def __eq__(self, other):	1✔
101	if self is other:	1✔
102	return True	1✔
103	if hasattr(other, "__len__") and len(self) != len(other):	1✔
104	return False	1✔
105	return seq_equals(self, other)	1✔
106
107	def __getitem__(self, item):	1✔
108	if isinstance(item, slice):	1✔
109	return PersistentVector(self._inner[item])	1✔
110	return self._inner[item]	1✔
111
112	def __hash__(self):	1✔
113	return hash(self._inner)	1✔
114
115	def __iter__(self):	1✔
116	yield from self._inner	1✔
117
118	def __len__(self):	1✔
119	return len(self._inner)	1✔
120
121	def __lt__(self, other):	1✔
122	if other is None: # pragma: no cover
123	return False
124	if not isinstance(other, PersistentVector):	1✔
125	return NotImplemented	1✔
126	if len(self) != len(other):	1✔
127	return len(self) < len(other)	1✔
128	return any(x < y for x, y in zip(self, other))	1✔
129
130	def _lrepr(self, **kwargs) -> str:	1✔
131	return _seq_lrepr(self._inner, "[", "]", meta=self._meta, **kwargs)	1✔
132
133	@property	1✔
134	def meta(self) -> Optional[IPersistentMap]:	1✔
135	return self._meta	1✔
136
137	def with_meta(self, meta: Optional[IPersistentMap]) -> "PersistentVector[T]":	1✔
138	return vector(self._inner, meta=meta)	1✔
139
140	def cons(self, *elems: T) -> "PersistentVector[T]": # type: ignore[override]	1✔
141	e = self._inner.evolver()	1✔
142	for elem in elems:	1✔
143	e.append(elem)	1✔
144	return PersistentVector(e.persistent(), meta=self.meta)	1✔
145
146	def assoc(self, *kvs: T) -> "PersistentVector[T]":	1✔
147	return PersistentVector(self._inner.mset(*kvs)) # type: ignore[arg-type]	1✔
148
149	def contains(self, k):	1✔
150	return 0 <= k < len(self._inner)	1✔
151
152	def entry(self, k):	1✔
153	try:	1✔
154	return MapEntry.of(k, self._inner[k])	1✔
155	except IndexError:	1✔
156	return None	1✔
157
158	def val_at(self, k, default=None):	1✔
159	try:	1✔
160	return self._inner[k]	1✔
161	except IndexError:	1✔
162	return default	1✔
163
164	@staticmethod	1✔
165	def empty() -> "PersistentVector[T]":	1✔
166	return EMPTY	1✔
167
168	def seq(self) -> Optional[ISeq[T]]: # type: ignore[override]	1✔
169	if len(self._inner) == 0:	1✔
170	return None	1✔
171	return sequence(self)	1✔
172
173	def peek(self) -> Optional[T]:	1✔
174	if len(self) == 0:	1✔
175	return None	1✔
176	return self[-1]	1✔
177
178	def pop(self) -> "PersistentVector[T]":	1✔
179	if len(self) == 0:	1✔
180	raise IndexError("Cannot pop an empty vector")	1✔
181	return self[:-1]	1✔
182
183	def rseq(self) -> ISeq[T]:	1✔
184	return sequence(reversed(self))	1✔
185
186	def to_transient(self) -> TransientVector:	1✔
187	return TransientVector(self._inner.evolver())	1✔
188
189
190	K = TypeVar("K")	1✔
191	V = TypeVar("V")	1✔
192
193
194	class MapEntry(IMapEntry[K, V], PersistentVector[Union[K, V]]):	1✔
195	__slots__ = ()	1✔
196
197	def __init__(self, wrapped: "PVector[Union[K, V]]") -> None:	1✔
198	try:	1✔
199	if not len(wrapped) == 2:	1✔
200	raise ValueError("Vector arg to map conj must be a pair")	1✔
201	except TypeError as e:	1✔
202	raise TypeError(f"Cannot make map entry from {type(wrapped)}") from e	×
203
204	super().__init__(wrapped)	1✔
205
206	@property	1✔
207	def key(self) -> K:	1✔
208	return self[0]	1✔
209
210	@property	1✔
211	def value(self) -> V:	1✔
212	return self[1]	1✔
213
214	@staticmethod	1✔
215	def of(k: K, v: V) -> "MapEntry[K, V]":	1✔
216	return MapEntry(pvector([k, v]))	1✔
217
218	@staticmethod	1✔
219	def from_vec(v: Sequence[Union[K, V]]) -> "MapEntry[K, V]":	1✔
220	return MapEntry(pvector(v))	1✔
221
222
223	EMPTY: PersistentVector = PersistentVector(pvector(()))	1✔
224
225
226	def vector(	1✔
227	members: Iterable[T], meta: Optional[IPersistentMap] = None
228	) -> PersistentVector[T]:
229	"""Creates a new vector."""
230	return PersistentVector(pvector(members), meta=meta)	1✔
231
232
233	def v(*members: T, meta: Optional[IPersistentMap] = None) -> PersistentVector[T]:	1✔
234	"""Creates a new vector from members."""
235	return PersistentVector(pvector(members), meta=meta)	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