10972474990

Committed 21 Sep 2024 12:15PM UTC coverage: 98.825% (-0.07%) from 98.89%

Build # 10972474990

Build Type

Pull #1067

github

Committed by

web-flow

Commit Message

Merge 2e9930d6f into dd7d239de

Pull Request Pull Request #1067: Add fixture functions to `basilisp.test` #980

Run Details

1316 of 1323 branches covered (99.47%)

Branch coverage included in aggregate %.

8695 of 8807 relevant lines covered (98.73%)

0.99 hits per line

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

96.97

/src/basilisp/lang/seq.py

import functools
import threading
from typing import Callable, Iterable, Optional, TypeVar, overload

from basilisp.lang.interfaces import (
    IPersistentMap,
    ISeq,
    ISeqable,
    ISequential,
    IWithMeta,
)
from basilisp.util import Maybe

T = TypeVar("T")


class _EmptySequence(IWithMeta, ISequential, ISeq[T]):
    __slots__ = ("_meta",)

    def __init__(self, meta: Optional[IPersistentMap] = None):
        self._meta = meta

    def __repr__(self):
        return "()"

    def __bool__(self):
        return True

    def seq(self) -> Optional[ISeq[T]]:
        return None

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

    def with_meta(self, meta: Optional[IPersistentMap]) -> "_EmptySequence[T]":
        return _EmptySequence(meta=meta)

    @property
    def is_empty(self) -> bool:
        return True

    @property
    def first(self) -> Optional[T]:
        return None

    @property
    def rest(self) -> ISeq[T]:
        return self

    def cons(self, *elems: T) -> ISeq[T]:  # type: ignore[override]
        l: ISeq = self
        for elem in elems:
            l = Cons(elem, l)
        return l

    @staticmethod
    def empty():
        return EMPTY


EMPTY: ISeq = _EmptySequence()


class Cons(ISeq[T], ISequential, IWithMeta):
    __slots__ = ("_first", "_rest", "_meta")

    def __init__(
        self,
        first: T,
        seq: Optional[ISeq[T]] = None,
        meta: Optional[IPersistentMap] = None,
    ) -> None:
        self._first = first
        self._rest = Maybe(seq).or_else_get(EMPTY)
        self._meta = meta

    @property
    def is_empty(self) -> bool:
        return False

    @property
    def first(self) -> Optional[T]:
        return self._first

    @property
    def rest(self) -> ISeq[T]:
        return self._rest

    def cons(self, *elems: T) -> "Cons[T]":
        l = self
        for elem in elems:
            l = Cons(elem, l)
        return l

    @staticmethod
    def empty():
        return EMPTY

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

    def with_meta(self, meta: Optional[IPersistentMap]) -> "Cons[T]":
        return Cons(self._first, seq=self._rest, meta=meta)


LazySeqGenerator = Callable[[], Optional[ISeq[T]]]


class LazySeq(IWithMeta, ISequential, ISeq[T]):
    """LazySeqs are wrappers for delaying sequence computation. Create a LazySeq
    with a function that can either return None or a Seq. If a Seq is returned,
    the LazySeq is a proxy to that Seq.

    Callers should never provide the `seq` argument -- this is provided only to
    support `with_meta` returning a new LazySeq instance."""

    __slots__ = ("_gen", "_obj", "_seq", "_lock", "_meta")

    def __init__(
        self,
        gen: Optional[LazySeqGenerator],
        seq: Optional[ISeq[T]] = None,
        *,
        meta: Optional[IPersistentMap] = None,
    ) -> None:
        self._gen: Optional[LazySeqGenerator] = gen
        self._obj: Optional[ISeq[T]] = None
        self._seq: Optional[ISeq[T]] = seq
        self._lock = threading.RLock()
        self._meta = meta

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

    def with_meta(self, meta: Optional[IPersistentMap]) -> "LazySeq[T]":
        return LazySeq(None, seq=self.seq(), meta=meta)

    # LazySeqs have a fairly complex inner state, in spite of the simple interface.
    # Calls from Basilisp code should be providing the only generator seed function.
    # Calls to `(seq ...)` cause the LazySeq to cache the generator function locally
    # (as explained in _compute_seq), clear the _gen attribute, and cache the results
    # of that generator function call as _obj. _obj may be None, some other ISeq, or
    # perhaps another LazySeq. Finally, the LazySeq attempts to consume all returned
    # LazySeq objects before calling `(seq ...)` on the result, which is cached in the
    # _seq attribute.

    def _compute_seq(self) -> Optional[ISeq[T]]:
        if self._gen is not None:
            # This local caching of the generator function and clearing of self._gen
            # is absolutely critical for supporting co-recursive lazy sequences.
            #
            # The original example that prompted this change is below:
            #
            #   (def primes (remove
            #                (fn [x] (some #(zero? (mod x %)) primes))
            #                (iterate inc 2)))
            #
            #   (take 5 primes)  ;; => stack overflow
            #
            # If we don't clear self._gen, each successive call to (some ... primes)
            # will end up forcing the primes LazySeq object to call self._gen, rather
            # than caching the results, allowing examination of the partial seq
            # computed up to that point.
            gen = self._gen
            self._gen = None
            self._obj = gen()
        return self._obj if self._obj is not None else self._seq

    def seq(self) -> Optional[ISeq[T]]:
        with self._lock:
            self._compute_seq()
            if self._obj is not None:
                o = self._obj
                self._obj = None
                # Consume any additional lazy sequences returned immediately, so we
                # have a "real" concrete sequence to proxy to.
                #
                # The common idiom with LazySeqs is to return
                # (cons value (lazy-seq ...)) from the generator function, so this will
                # only result in evaluating away instances where _another_ LazySeq is
                # returned rather than a cons cell with a concrete first value. This
                # loop will not consume the LazySeq in the rest position of the cons.
                while isinstance(o, LazySeq):
                    o = o._compute_seq()  # type: ignore
                self._seq = to_seq(o)
            return self._seq

    @property
    def is_empty(self) -> bool:
        return self.seq() is None

    @property
    def first(self) -> Optional[T]:
        if self.is_empty:
            return None
        return self.seq().first  # type: ignore[union-attr]

    @property
    def rest(self) -> "ISeq[T]":
        if self.is_empty:
            return EMPTY
        return self.seq().rest  # type: ignore[union-attr]

    def cons(self, *elems: T) -> ISeq[T]:  # type: ignore[override]
        l: ISeq = self
        for elem in elems:
            l = Cons(elem, l)
        return l

    @property
    def is_realized(self):
        with self._lock:
            return self._gen is None

    @staticmethod
    def empty():
        return EMPTY


def sequence(s: Iterable[T]) -> ISeq[T]:
    """Create a Sequence from Iterable s."""
    i = iter(s)

    def _next_elem() -> ISeq[T]:
        try:
            e = next(i)
        except StopIteration:
            return EMPTY
        else:
            return Cons(e, LazySeq(_next_elem))

    return LazySeq(_next_elem)


@overload
def _seq_or_nil(s: None) -> None: ...


@overload
def _seq_or_nil(s: ISeq) -> Optional[ISeq]: ...


def _seq_or_nil(s):
    """Return None if a ISeq is empty, the ISeq otherwise."""
    if s is None or s.is_empty:
        return None
    return s


@functools.singledispatch
def to_seq(o) -> Optional[ISeq]:
    """Coerce the argument o to a ISeq. If o is None, return None."""
    return _seq_or_nil(sequence(o))


@to_seq.register(type(None))
def _to_seq_none(_) -> None:
    return None


@to_seq.register(ISeq)
def _to_seq_iseq(o: ISeq) -> Optional[ISeq]:
    return _seq_or_nil(o)


@to_seq.register(LazySeq)
def _to_seq_lazyseq(o: LazySeq) -> Optional[ISeq]:
    # Force evaluation of the LazySeq by calling o.seq() directly.
    return o.seq()


@to_seq.register(ISeqable)
def _to_seq_iseqable(o: ISeqable) -> Optional[ISeq]:
    return _seq_or_nil(o.seq())

1	import functools	1✔
2	import threading	1✔
3	from typing import Callable, Iterable, Optional, TypeVar, overload	1✔
4
5	from basilisp.lang.interfaces import (	1✔
6	IPersistentMap,
7	ISeq,
8	ISeqable,
9	ISequential,
10	IWithMeta,
11	)
12	from basilisp.util import Maybe	1✔
13
14	T = TypeVar("T")	1✔
15
16
17	class _EmptySequence(IWithMeta, ISequential, ISeq[T]):	1✔
18	__slots__ = ("_meta",)	1✔
19
20	def __init__(self, meta: Optional[IPersistentMap] = None):	1✔
21	self._meta = meta	1✔
22
23	def __repr__(self):
24	return "()"
25
26	def __bool__(self):	1✔
27	return True	1✔
28
29	def seq(self) -> Optional[ISeq[T]]:	1✔
30	return None	×
31
32	@property	1✔
33	def meta(self) -> Optional[IPersistentMap]:	1✔
34	return self._meta	×
35
36	def with_meta(self, meta: Optional[IPersistentMap]) -> "_EmptySequence[T]":	1✔
37	return _EmptySequence(meta=meta)	×
38
39	@property	1✔
40	def is_empty(self) -> bool:	1✔
41	return True	1✔
42
43	@property	1✔
44	def first(self) -> Optional[T]:	1✔
45	return None	1✔
46
47	@property	1✔
48	def rest(self) -> ISeq[T]:	1✔
49	return self	×
50
51	def cons(self, *elems: T) -> ISeq[T]: # type: ignore[override]	1✔
52	l: ISeq = self	1✔
53	for elem in elems:	1✔
54	l = Cons(elem, l)	1✔
55	return l	1✔
56
57	@staticmethod	1✔
58	def empty():	1✔
59	return EMPTY	×
60
61
62	EMPTY: ISeq = _EmptySequence()	1✔
63
64
65	class Cons(ISeq[T], ISequential, IWithMeta):	1✔
66	__slots__ = ("_first", "_rest", "_meta")	1✔
67
68	def __init__(	1✔
69	self,
70	first: T,
71	seq: Optional[ISeq[T]] = None,
72	meta: Optional[IPersistentMap] = None,
73	) -> None:
74	self._first = first	1✔
75	self._rest = Maybe(seq).or_else_get(EMPTY)	1✔
76	self._meta = meta	1✔
77
78	@property	1✔
79	def is_empty(self) -> bool:	1✔
80	return False	1✔
81
82	@property	1✔
83	def first(self) -> Optional[T]:	1✔
84	return self._first	1✔
85
86	@property	1✔
87	def rest(self) -> ISeq[T]:	1✔
88	return self._rest	1✔
89
90	def cons(self, *elems: T) -> "Cons[T]":	1✔
91	l = self	1✔
92	for elem in elems:	1✔
93	l = Cons(elem, l)	1✔
94	return l	1✔
95
96	@staticmethod	1✔
97	def empty():	1✔
98	return EMPTY	1✔
99
100	@property	1✔
101	def meta(self) -> Optional[IPersistentMap]:	1✔
102	return self._meta	1✔
103
104	def with_meta(self, meta: Optional[IPersistentMap]) -> "Cons[T]":	1✔
105	return Cons(self._first, seq=self._rest, meta=meta)	1✔
106
107
108	LazySeqGenerator = Callable[[], Optional[ISeq[T]]]	1✔
109
110
111	class LazySeq(IWithMeta, ISequential, ISeq[T]):	1✔
112	"""LazySeqs are wrappers for delaying sequence computation. Create a LazySeq
113	with a function that can either return None or a Seq. If a Seq is returned,
114	the LazySeq is a proxy to that Seq.
115
116	Callers should never provide the `seq` argument -- this is provided only to
117	support `with_meta` returning a new LazySeq instance."""
118
119	__slots__ = ("_gen", "_obj", "_seq", "_lock", "_meta")	1✔
120
121	def __init__(	1✔
122	self,
123	gen: Optional[LazySeqGenerator],
124	seq: Optional[ISeq[T]] = None,
125	*,
126	meta: Optional[IPersistentMap] = None,
127	) -> None:
128	self._gen: Optional[LazySeqGenerator] = gen	1✔
129	self._obj: Optional[ISeq[T]] = None	1✔
130	self._seq: Optional[ISeq[T]] = seq	1✔
131	self._lock = threading.RLock()	1✔
132	self._meta = meta	1✔
133
134	@property	1✔
135	def meta(self) -> Optional[IPersistentMap]:	1✔
136	return self._meta	1✔
137
138	def with_meta(self, meta: Optional[IPersistentMap]) -> "LazySeq[T]":	1✔
139	return LazySeq(None, seq=self.seq(), meta=meta)	1✔
140
141	# LazySeqs have a fairly complex inner state, in spite of the simple interface.
142	# Calls from Basilisp code should be providing the only generator seed function.
143	# Calls to `(seq ...)` cause the LazySeq to cache the generator function locally
144	# (as explained in _compute_seq), clear the _gen attribute, and cache the results
145	# of that generator function call as _obj. _obj may be None, some other ISeq, or
146	# perhaps another LazySeq. Finally, the LazySeq attempts to consume all returned
147	# LazySeq objects before calling `(seq ...)` on the result, which is cached in the
148	# _seq attribute.
149
150	def _compute_seq(self) -> Optional[ISeq[T]]:	1✔
151	if self._gen is not None:	1✔
152	# This local caching of the generator function and clearing of self._gen
153	# is absolutely critical for supporting co-recursive lazy sequences.
154	#
155	# The original example that prompted this change is below:
156	#
157	# (def primes (remove
158	# (fn [x] (some #(zero? (mod x %)) primes))
159	# (iterate inc 2)))
160	#
161	# (take 5 primes) ;; => stack overflow
162	#
163	# If we don't clear self._gen, each successive call to (some ... primes)
164	# will end up forcing the primes LazySeq object to call self._gen, rather
165	# than caching the results, allowing examination of the partial seq
166	# computed up to that point.
167	gen = self._gen	1✔
168	self._gen = None	1✔
169	self._obj = gen()	1✔
170	return self._obj if self._obj is not None else self._seq	1✔
171
172	def seq(self) -> Optional[ISeq[T]]:	1✔
173	with self._lock:	1✔
174	self._compute_seq()	1✔
175	if self._obj is not None:	1✔
176	o = self._obj	1✔
177	self._obj = None	1✔
178	# Consume any additional lazy sequences returned immediately, so we
179	# have a "real" concrete sequence to proxy to.
180	#
181	# The common idiom with LazySeqs is to return
182	# (cons value (lazy-seq ...)) from the generator function, so this will
183	# only result in evaluating away instances where _another_ LazySeq is
184	# returned rather than a cons cell with a concrete first value. This
185	# loop will not consume the LazySeq in the rest position of the cons.
186	while isinstance(o, LazySeq):	1✔
187	o = o._compute_seq() # type: ignore	1✔
188	self._seq = to_seq(o)	1✔
189	return self._seq	1✔
190
191	@property	1✔
192	def is_empty(self) -> bool:	1✔
193	return self.seq() is None	1✔
194
195	@property	1✔
196	def first(self) -> Optional[T]:	1✔
197	if self.is_empty:	1✔
198	return None	1✔
199	return self.seq().first # type: ignore[union-attr]	1✔
200
201	@property	1✔
202	def rest(self) -> "ISeq[T]":	1✔
203	if self.is_empty:	1✔
204	return EMPTY	1✔
205	return self.seq().rest # type: ignore[union-attr]	1✔
206
207	def cons(self, *elems: T) -> ISeq[T]: # type: ignore[override]	1✔
208	l: ISeq = self	1✔
209	for elem in elems:	1✔
210	l = Cons(elem, l)	1✔
211	return l	1✔
212
213	@property	1✔
214	def is_realized(self):	1✔
215	with self._lock:	1✔
216	return self._gen is None	1✔
217
218	@staticmethod	1✔
219	def empty():	1✔
220	return EMPTY	1✔
221
222
223	def sequence(s: Iterable[T]) -> ISeq[T]:	1✔
224	"""Create a Sequence from Iterable s."""
225	i = iter(s)	1✔
226
227	def _next_elem() -> ISeq[T]:	1✔
228	try:	1✔
229	e = next(i)	1✔
230	except StopIteration:	1✔
231	return EMPTY	1✔
232	else:
233	return Cons(e, LazySeq(_next_elem))	1✔
234
235	return LazySeq(_next_elem)	1✔
236
237
238	@overload	1✔
239	def _seq_or_nil(s: None) -> None: ...	1✔
240
241
242	@overload	1✔
243	def _seq_or_nil(s: ISeq) -> Optional[ISeq]: ...	1✔
244
245
246	def _seq_or_nil(s):	1✔
247	"""Return None if a ISeq is empty, the ISeq otherwise."""
248	if s is None or s.is_empty:	1✔
249	return None	1✔
250	return s	1✔
251
252
253	@functools.singledispatch	1✔
254	def to_seq(o) -> Optional[ISeq]:	1✔
255	"""Coerce the argument o to a ISeq. If o is None, return None."""
256	return _seq_or_nil(sequence(o))	1✔
257
258
259	@to_seq.register(type(None))	1✔
260	def _to_seq_none(_) -> None:	1✔
261	return None	1✔
262
263
264	@to_seq.register(ISeq)	1✔
265	def _to_seq_iseq(o: ISeq) -> Optional[ISeq]:	1✔
266	return _seq_or_nil(o)	1✔
267
268
269	@to_seq.register(LazySeq)	1✔
270	def _to_seq_lazyseq(o: LazySeq) -> Optional[ISeq]:	1✔
271	# Force evaluation of the LazySeq by calling o.seq() directly.
272	return o.seq()	1✔
273
274
275	@to_seq.register(ISeqable)	1✔
276	def _to_seq_iseqable(o: ISeqable) -> Optional[ISeq]:	1✔
277	return _seq_or_nil(o.seq())	1✔

basilisp-lang / basilisp / 10972474990

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