feat : Added new generator - csv_row_reader, directory_watcher and sliding_window.

Author: manas-shinde

generators/csv_row_reader.py

@@ -0,0 +1,9 @@
+import csv
+
+
+def csv_row_reader(filepath):
+    """Yields rows from a CSV file as dictionaries (using header row)."""
+    with open(filepath, newline='', encoding='utf-8') as f:
+        reader = csv.DictReader(f)
+        for row in reader:
+            yield row

generators/directory_watcher.py

@@ -0,0 +1,17 @@
+import os
+import time
+
+
+def directory_watcher(path, interval=1.0):
+    """
+    Yields new filenames in a directory as they appear.
+    """
+    seen = set(os.listdir(path))
+
+    while True:
+        time.sleep(interval)
+        current = set(os.listdir(path))
+        new_files = current - seen
+        for file in new_files:
+            yield file
+        seen = current

generators/sliding_window.py

@@ -0,0 +1,12 @@
+from collections import deque
+
+
+def sliding_window(iterable, size):
+    """Yields sliding windows of the given size from an iterable."""
+    it = iter(iterable)
+    window = deque(maxlen=size)
+
+    for item in it:
+        window.append(item)
+        if len(window) == size:
+            yield tuple(window)

tests/test_csv_row_reader.py

@@ -0,0 +1,16 @@
+import tempfile
+import csv
+from generators.csv_row_reader import csv_row_reader
+
+
+def test_csv_row_reader():
+    with tempfile.NamedTemporaryFile(mode='w+', newline='', delete=False) as f:
+        writer = csv.DictWriter(f, fieldnames=["id", "name"])
+        writer.writeheader()
+        writer.writerow({"id": "1", "name": "Alice"})
+        writer.writerow({"id": "2", "name": "Bob"})
+        path = f.name
+
+    rows = list(csv_row_reader(path))
+    assert len(rows) == 2
+    assert rows[0]["name"] == "Alice"

tests/test_sliding_window.py

@@ -0,0 +1,6 @@
+from generators.sliding_window import sliding_window
+
+
+def test_sliding_window():
+    result = list(sliding_window(range(5), 3))
+    assert result == [(0, 1, 2), (1, 2, 3), (2, 3, 4)]