cython

1
from cython.cimports import cqueue
2
from cython import cast
3

4
@cython.cclass
5
class Queue:
6
    """A queue class for C integer values.
7

8
    >>> q = Queue()
9
    >>> q.append(5)
10
    >>> q.peek()
11
    5
12
    >>> q.pop()
13
    5
14
    """
15
    _c_queue = cython.declare(cython.pointer(cqueue.Queue))
16
    def __cinit__(self):
17
        self._c_queue = cqueue.queue_new()
18
        if self._c_queue is cython.NULL:
19
            raise MemoryError()
20

21
    def __dealloc__(self):
22
        if self._c_queue is not cython.NULL:
23
            cqueue.queue_free(self._c_queue)
24

25
    @cython.ccall
26
    def append(self, value: cython.int):
27
        if not cqueue.queue_push_tail(self._c_queue,
28
                cast(cython.p_void, cast(cython.Py_ssize_t, value))):
29
            raise MemoryError()
30

31
    # The `cpdef` feature is obviously not available for the original "extend()"
32
    # method, as the method signature is incompatible with Python argument
33
    # types (Python does not have pointers).  However, we can rename
34
    # the C-ish "extend()" method to e.g. "extend_ints()", and write
35
    # a new "extend()" method that provides a suitable Python interface by
36
    # accepting an arbitrary Python iterable.
37
    @cython.ccall
38
    def extend(self, values):
39
        for value in values:
40
            self.append(value)
41

42
    @cython.cfunc
43
    def extend_ints(self, values: cython.p_int, count: cython.size_t):
44
        value: cython.int
45
        for value in values[:count]:  # Slicing pointer to limit the iteration boundaries.
46
            self.append(value)
47

48
    @cython.ccall
49
    @cython.exceptval(-1, check=True)
50
    def peek(self) -> cython.int:
51
        value: cython.int = cast(cython.Py_ssize_t, cqueue.queue_peek_head(self._c_queue))
52

53
        if value == 0:
54
            # this may mean that the queue is empty,
55
            # or that it happens to contain a 0 value
56
            if cqueue.queue_is_empty(self._c_queue):
57
                raise IndexError("Queue is empty")
58
        return value
59

60
    @cython.ccall
61
    @cython.exceptval(-1, check=True)
62
    def pop(self) -> cython.int:
63
        if cqueue.queue_is_empty(self._c_queue):
64
            raise IndexError("Queue is empty")
65
        return cast(cython.Py_ssize_t, cqueue.queue_pop_head(self._c_queue))
66

67
    def __bool__(self):
68
        return not cqueue.queue_is_empty(self._c_queue)
69