cython

1
cimport cqueue
2

3

4

5
cdef 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
    cdef cqueue.Queue* _c_queue
16
    def __cinit__(self):
17
        self._c_queue = cqueue.queue_new()
18
        if self._c_queue is NULL:
19
            raise MemoryError()
20

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

25

26
    cpdef append(self, int value):
27
        if not cqueue.queue_push_tail(self._c_queue,
28
                                      <void*> <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

38
    cpdef extend(self, values):
39
        for value in values:
40
            self.append(value)
41

42

43
    cdef extend_ints(self, int* values, size_t count):
44
        cdef int value
45
        for value in values[:count]:  # Slicing pointer to limit the iteration boundaries.
46
            self.append(value)
47

48

49

50
    cpdef int peek(self) except? -1:
51
        cdef int value = <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

61

62
    cpdef int pop(self) except? -1:
63
        if cqueue.queue_is_empty(self._c_queue):
64
            raise IndexError("Queue is empty")
65
        return <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