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,43 @@
+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:
+            # Break exception chain to hide implementation details
+            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 __iter__(self):
+        yield from self._items
+
+    def __reversed__(self):
+        yield from reversed(self._items)
+
+    def __repr__(self):
+        return f"Queue({list(self._items)})"
+
+
+if __name__ == "__main__":
+    queue = Queue()
+    print("Enqueueing items with '.enqueue()'...")
+    queue.enqueue(1)
+    queue.enqueue(2)
+    queue.enqueue(3)
+    print(queue)
+    print("Dequeueing one item with '.dequeue()'...")
+    print(queue.dequeue())
+    print(queue)

python-deque/page_history.py

@@ -0,0 +1,18 @@
+from collections import deque
+
+page_history = deque(maxlen=3)
+
+urls = ("https://google.com", "https://yahoo.com", "https://www.bing.com")
+
+for url in urls:
+    page_history.appendleft(url)
+
+print(page_history)
+
+page_history.appendleft("https://youtube.com")
+
+print(page_history)
+
+page_history.appendleft("https://facebook.com")
+
+print(page_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/tail.py

@@ -0,0 +1,13 @@
+from collections import deque
+
+
+def tail(filename, lines=10):
+    try:
+        with open(filename) as file:
+            return deque(file, maxlen=lines)
+    except OSError as error:
+        print(f'Opening file "{filename}" failed with error: {error}')
+
+
+if __name__ == "__main__":
+    print(tail("./README.md"))

python-deque/time_append.py

@@ -0,0 +1,25 @@
+from collections import deque
+from time import perf_counter
+
+TIMES = 10_000
+NANOSECONDS_PER_SECOND = 1e9
+a_list = []
+a_deque = deque()
+
+
+def average_time(func, times):
+    total = 0.0
+    for i in range(times):
+        start = perf_counter()
+        func(i)
+        # Convert to ns to improve readability
+        total += (perf_counter() - start) * NANOSECONDS_PER_SECOND
+    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,25 @@
+from collections import deque
+from time import perf_counter
+
+TIMES = 10_000
+NANOSECONDS_PER_SECOND = 1e9
+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()
+        # Convert to ns to improve readability
+        total += (perf_counter() - start) * NANOSECONDS_PER_SECOND
+    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(0)     {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,33 @@
+from collections import deque
+from time import perf_counter
+
+TIMES = 10_000
+MICROSECONDS_PER_SECOND = 1e6
+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()
+        # Convert to μs to improve readability
+        total += (perf_counter() - start) * MICROSECONDS_PER_SECOND
+    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")