summaryrefslogtreecommitdiff
path: root/docs/examples/tutorial/clibraries/queue3.py
diff options
context:
space:
mode:
Diffstat (limited to 'docs/examples/tutorial/clibraries/queue3.py')
-rw-r--r--docs/examples/tutorial/clibraries/queue3.py68
1 files changed, 68 insertions, 0 deletions
diff --git a/docs/examples/tutorial/clibraries/queue3.py b/docs/examples/tutorial/clibraries/queue3.py
new file mode 100644
index 000000000..79f341254
--- /dev/null
+++ b/docs/examples/tutorial/clibraries/queue3.py
@@ -0,0 +1,68 @@
+from cython.cimports import cqueue
+from cython import cast
+
+@cython.cclass
+class Queue:
+ """A queue class for C integer values.
+
+ >>> q = Queue()
+ >>> q.append(5)
+ >>> q.peek()
+ 5
+ >>> q.pop()
+ 5
+ """
+ _c_queue = cython.declare(cython.pointer(cqueue.Queue))
+ def __cinit__(self):
+ self._c_queue = cqueue.queue_new()
+ if self._c_queue is cython.NULL:
+ raise MemoryError()
+
+ def __dealloc__(self):
+ if self._c_queue is not cython.NULL:
+ cqueue.queue_free(self._c_queue)
+
+ @cython.ccall
+ def append(self, value: cython.int):
+ if not cqueue.queue_push_tail(self._c_queue,
+ cast(cython.p_void, cast(cython.Py_ssize_t, value))):
+ raise MemoryError()
+
+ # The `cpdef` feature is obviously not available for the original "extend()"
+ # method, as the method signature is incompatible with Python argument
+ # types (Python does not have pointers). However, we can rename
+ # the C-ish "extend()" method to e.g. "extend_ints()", and write
+ # a new "extend()" method that provides a suitable Python interface by
+ # accepting an arbitrary Python iterable.
+ @cython.ccall
+ def extend(self, values):
+ for value in values:
+ self.append(value)
+
+ @cython.cfunc
+ def extend_ints(self, values: cython.p_int, count: cython.size_t):
+ value: cython.int
+ for value in values[:count]: # Slicing pointer to limit the iteration boundaries.
+ self.append(value)
+
+ @cython.ccall
+ @cython.exceptval(-1, check=True)
+ def peek(self) -> cython.int:
+ value: cython.int = cast(cython.Py_ssize_t, cqueue.queue_peek_head(self._c_queue))
+
+ if value == 0:
+ # this may mean that the queue is empty,
+ # or that it happens to contain a 0 value
+ if cqueue.queue_is_empty(self._c_queue):
+ raise IndexError("Queue is empty")
+ return value
+
+ @cython.ccall
+ @cython.exceptval(-1, check=True)
+ def pop(self) -> cython.int:
+ if cqueue.queue_is_empty(self._c_queue):
+ raise IndexError("Queue is empty")
+ return cast(cython.Py_ssize_t, cqueue.queue_pop_head(self._c_queue))
+
+ def __bool__(self):
+ return not cqueue.queue_is_empty(self._c_queue)