5697667142

Committed 29 Jul 2023 01:09AM CUT 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

98.36

/src/basilisp/lang/queue.py

from typing import Optional, TypeVar

from pyrsistent import PDeque, pdeque  # noqa # pylint: disable=unused-import

from basilisp.lang.interfaces import (
    ILispObject,
    IPersistentList,
    IPersistentMap,
    ISeq,
    IWithMeta,
    seq_equals,
)
from basilisp.lang.obj import seq_lrepr as _seq_lrepr
from basilisp.lang.seq import sequence

T = TypeVar("T")


class PersistentQueue(IPersistentList[T], IWithMeta, ILispObject):
    """Basilisp Queue. Delegates internally to a pyrsistent.PDeque object.

    Do not instantiate directly. Instead use the q() and queue() factory
    methods below."""

    __slots__ = ("_inner", "_meta")

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

    def __bool__(self):
        return True

    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 __hash__(self):
        return hash(self._inner)

    def __iter__(self):
        yield from self._inner

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

    def _lrepr(self, **kwargs) -> str:
        return _seq_lrepr(self._inner, "#queue (", ")", meta=self._meta, **kwargs)

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

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

    def cons(self, *elems: T) -> "PersistentQueue[T]":
        return PersistentQueue(self._inner.extend(elems))

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

    def peek(self):
        try:
            return self._inner.left
        except IndexError:
            return None

    def pop(self) -> "PersistentQueue[T]":
        if len(self._inner) == 0:
            raise IndexError("Cannot pop an empty queue")
        return PersistentQueue(self._inner.popleft())

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


EMPTY: PersistentQueue = PersistentQueue(pdeque())


def queue(members, meta=None) -> PersistentQueue:
    """Creates a new queue."""
    return PersistentQueue(pdeque(iterable=members), meta=meta)


def q(*members, meta=None) -> PersistentQueue:  # noqa
    """Creates a new queue from members."""
    return PersistentQueue(pdeque(iterable=members), meta=meta)

1	from typing import Optional, TypeVar	1✔
2
3	from pyrsistent import PDeque, pdeque # noqa # pylint: disable=unused-import	1✔
4
5	from basilisp.lang.interfaces import (	1✔
6	ILispObject,
7	IPersistentList,
8	IPersistentMap,
9	ISeq,
10	IWithMeta,
11	seq_equals,
12	)
13	from basilisp.lang.obj import seq_lrepr as _seq_lrepr	1✔
14	from basilisp.lang.seq import sequence	1✔
15
16	T = TypeVar("T")	1✔
17
18
19	class PersistentQueue(IPersistentList[T], IWithMeta, ILispObject):	1✔
20	"""Basilisp Queue. Delegates internally to a pyrsistent.PDeque object.
21
22	Do not instantiate directly. Instead use the q() and queue() factory
23	methods below."""
24
25	__slots__ = ("_inner", "_meta")	1✔
26
27	def __init__(self, wrapped: "PDeque[T]", meta=None) -> None:	1✔
28	self._inner = wrapped	1✔
29	self._meta = meta	1✔
30
31	def __bool__(self):	1✔
32	return True	1✔
33
34	def __eq__(self, other):	1✔
35	if self is other:	1✔
36	return True	1✔
37	if hasattr(other, "__len__") and len(self) != len(other):	1✔
38	return False	1✔
39	return seq_equals(self, other)	1✔
40
41	def __hash__(self):	1✔
42	return hash(self._inner)	×
43
44	def __iter__(self):	1✔
45	yield from self._inner	1✔
46
47	def __len__(self):	1✔
48	return len(self._inner)	1✔
49
50	def _lrepr(self, **kwargs) -> str:	1✔
51	return _seq_lrepr(self._inner, "#queue (", ")", meta=self._meta, **kwargs)	1✔
52
53	@property	1✔
54	def meta(self) -> Optional[IPersistentMap]:	1✔
55	return self._meta	1✔
56
57	def with_meta(self, meta: Optional[IPersistentMap]) -> "PersistentQueue":	1✔
58	return queue(self._inner, meta=meta)	1✔
59
60	def cons(self, *elems: T) -> "PersistentQueue[T]":	1✔
61	return PersistentQueue(self._inner.extend(elems))	1✔
62
63	@staticmethod	1✔
64	def empty() -> "PersistentQueue":	1✔
65	return EMPTY	1✔
66
67	def peek(self):	1✔
68	try:	1✔
69	return self._inner.left	1✔
70	except IndexError:	1✔
71	return None	1✔
72
73	def pop(self) -> "PersistentQueue[T]":	1✔
74	if len(self._inner) == 0:	1✔
75	raise IndexError("Cannot pop an empty queue")	1✔
76	return PersistentQueue(self._inner.popleft())	1✔
77
78	def seq(self) -> Optional[ISeq[T]]:	1✔
79	if len(self._inner) == 0:	1✔
80	return None	1✔
81	return sequence(self)	1✔
82
83
84	EMPTY: PersistentQueue = PersistentQueue(pdeque())	1✔
85
86
87	def queue(members, meta=None) -> PersistentQueue:	1✔
88	"""Creates a new queue."""
89	return PersistentQueue(pdeque(iterable=members), meta=meta)	1✔
90
91
92	def q(*members, meta=None) -> PersistentQueue: # noqa	1✔
93	"""Creates a new queue from members."""
94	return PersistentQueue(pdeque(iterable=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