Sample code for the article on deque

python-deque/README.md

@@ -0,0 +1,3 @@
+# Python's deque: Implement Efficient Queues and Stacks
+
+This folder provides the code examples for the Real Python tutorial [Python's deque: Implement Efficient Queues and Stacks](https://realpython.com/python-deque/).

python-deque/custom_queue.py

@@ -0,0 +1,30 @@
+from collections import deque
+
+
+class Queue:
+    def __init__(self):
+        self._items = deque()
+
+    def enqueue(self, item):
+        self._items.append(item)
+
+    def dequeue(self):
+        try:
+            return self._items.popleft()
+        except IndexError:
+            raise IndexError("dequeue from an empty queue") from None
+
+    def __len__(self):
+        return len(self._items)
+
+    def __contains__(self, item):
+        return item in self._items
+
+    def __str__(self):
+        return f"Queue({list(self._items)})"
+
+    def __iter__(self):
+        yield from self._items
+
+    def __reversed__(self):
+        yield from reversed(self._items)

python-deque/page.py

@@ -0,0 +1,18 @@
+from collections import deque
+
+doc_history = deque(maxlen=3)
+
+urls = ("https://google.com", "https://yahoo.com", "https://www.bing.com")
+
+for url in urls:
+    doc_history.appendleft(url)
+
+print(doc_history)
+
+doc_history.appendleft("https://youtube.com")
+
+print(doc_history)
+
+doc_history.appendleft("https://facebook.com")
+
+print(doc_history)

python-deque/producer_consumer.py

@@ -0,0 +1,42 @@
+import logging
+import random
+import threading
+import time
+from collections import deque
+
+logging.basicConfig(level=logging.INFO, format="%(message)s")
+
+
+def wait_seconds(mins, maxs):
+    time.sleep(mins + random.random() * (maxs - mins))
+
+
+def produce(queue, size):
+    while True:
+        if len(queue) < size:
+            value = random.randint(0, 9)
+            queue.append(value)
+            logging.info("Produced: %d -> %s", value, str(queue))
+        else:
+            logging.info("Queue is saturated")
+        wait_seconds(0.1, 0.5)
+
+
+def consume(queue):
+    while True:
+        try:
+            value = queue.popleft()
+        except IndexError:
+            logging.info("Queue is empty")
+        else:
+            logging.info("Consumed: %d -> %s", value, str(queue))
+        wait_seconds(0.2, 0.7)
+
+
+logging.info("Starting Threads...\n")
+logging.info("Press Ctrl+C to interrupt the execution\n")
+
+shared_queue = deque()
+
+threading.Thread(target=produce, args=(shared_queue, 10)).start()
+threading.Thread(target=consume, args=(shared_queue,)).start()

python-deque/rotate.py

@@ -0,0 +1,21 @@
+from collections import deque
+
+
+def slice_deque(deck, *, start, stop):
+    slice = deque()
+    temp = deque()
+    for i in range(stop, start, -1):
+        deck.rotate(i)
+        item = deck.popleft()
+        slice.appendleft(item)
+        temp.append(item)
+    deck.extend(temp)
+    # for i in range(start, stop):
+    #     deck.rotate(-i)
+
+    return slice
+
+
+d = deque([1, 2, 3, 4, 5])
+print(slice_deque(d, start=0, stop=3))
+print(d)

python-deque/tail.py

@@ -0,0 +1,9 @@
+from collections import deque
+
+
+def tail(filename, lines=10):
+    try:
+        with open(filename) as file:
+            return deque(file, lines)
+    except OSError as error:
+        print(f'Opening file "{filename}" failed with error: {error}')

python-deque/threads.py

@@ -0,0 +1,42 @@
+import logging
+import random
+import threading
+import time
+from collections import deque
+
+logging.basicConfig(level=logging.INFO, format="%(message)s")
+
+
+def wait_seconds(mins, maxs):
+    time.sleep(mins + random.random() * (maxs - mins))
+
+
+def produce(queue, size):
+    while True:
+        if len(queue) < size:
+            value = random.randint(0, 9)
+            queue.append(value)
+            logging.info("Produced: %d -> %s", value, str(queue))
+        else:
+            logging.info("Queue is saturated")
+        wait_seconds(0.1, 0.5)
+
+
+def consume(queue):
+    while True:
+        try:
+            value = queue.popleft()
+        except IndexError:
+            logging.info("Queue is empty")
+        else:
+            logging.info("Consumed: %d -> %s", value, str(queue))
+        wait_seconds(0.2, 0.7)
+
+
+logging.info("Starting Threads...\n")
+logging.info("Press Ctrl+C to interrupt the execution\n")
+
+shared_queue = deque()
+
+threading.Thread(target=produce, args=(shared_queue, 10)).start()
+threading.Thread(target=consume, args=(shared_queue,)).start()

python-deque/time_append.py

@@ -0,0 +1,23 @@
+from collections import deque
+from time import perf_counter
+
+TIMES = 10_000
+a_list = []
+a_deque = deque()
+
+
+def average_time(func, times):
+    total = 0.0
+    for i in range(times):
+        start = perf_counter()
+        func(i)
+        total += (perf_counter() - start) * 1e9
+    return total / times
+
+
+list_time = average_time(lambda i: a_list.insert(0, i), TIMES)
+deque_time = average_time(lambda i: a_deque.appendleft(i), TIMES)
+gain = list_time / deque_time
+
+print(f"list.insert()      {list_time:.6} ns")
+print(f"deque.appendleft() {deque_time:.6} ns  ({gain:.6}x faster)")

python-deque/time_pop.py

@@ -0,0 +1,23 @@
+from collections import deque
+from time import perf_counter
+
+TIMES = 10000
+a_list = [1] * TIMES
+a_deque = deque(a_list)
+
+
+def average_time(func, times):
+    total = 0.0
+    for _ in range(times):
+        start = perf_counter()
+        func()
+        total += (perf_counter() - start) * 1e9
+    return total / times
+
+
+list_time = average_time(lambda: a_list.pop(0), TIMES)
+deque_time = average_time(lambda: a_deque.popleft(), TIMES)
+gain = list_time / deque_time
+
+print(f"list.pop()      {list_time:.6} ns")
+print(f"deque.popleft() {deque_time:.6} ns  ({gain:.6}x faster)")

python-deque/time_random_access.py

@@ -0,0 +1,31 @@
+from collections import deque
+from time import perf_counter
+
+TIMES = 10_000
+a_list = [1] * TIMES
+a_deque = deque(a_list)
+
+
+def average_time(func, times):
+    total = 0.0
+    for _ in range(times):
+        start = perf_counter()
+        func()
+        total += (perf_counter() - start) * 1e6
+    return total / times
+
+
+def time_it(sequence):
+    middle = len(sequence) // 2
+    sequence.insert(middle, "middle")
+    sequence[middle]
+    sequence.remove("middle")
+    del sequence[middle]
+
+
+list_time = average_time(lambda: time_it(a_list), TIMES)
+deque_time = average_time(lambda: time_it(a_deque), TIMES)
+gain = deque_time / list_time
+
+print(f"list  {list_time:.6} μs ({gain:.6}x faster)")
+print(f"deque {deque_time:.6} μs")