How to Mock a File Source Watch Directory in Python
Poll a directory for new files and log changes, simulating a watch directory for data ingestion patterns.
Python code
46 linesimport os
import time
from pathlib import Path
def watch_directory(dir_path: str, poll_interval: float = 1.0, max_iterations: int = 5):
"""
Mock a file-source watch directory by polling for changes.
Returns new files detected during each poll cycle.
"""
directory = Path(dir_path)
directory.mkdir(exist_ok=True)
seen = {p.name for p in directory.iterdir()}
print(f"Watching {directory} — initial files: {sorted(seen)}")
for iteration in range(1, max_iterations + 1):
time.sleep(poll_interval)
current = {p.name for p in directory.iterdir() if p.is_file()}
new_files = current - seen
if new_files:
print(f"Poll {iteration}: new files -> {sorted(new_files)}")
seen.update(new_files)
else:
print(f"Poll {iteration}: no changes")
return sorted(seen)
if __name__ == "__main__":
watch_dir = Path("mock_watch_dir")
demo_file = watch_dir / "demo.txt"
demo_file.write_text("watching demo content")
# Simulate a new file appearing mid-poll
import threading
def add_late_file():
time.sleep(2.5)
(watch_dir / "late_file.log").write_text("arrives later")
threading.Thread(target=add_late_file, daemon=True).start()
final_files = watch_directory(str(watch_dir), poll_interval=1.0, max_iterations=4)
print(f"Final tracked files: {final_files}")
demo_file.unlink(missing_ok=True)
(watch_dir / "late_file.log").unlink(missing_ok=True)
watch_dir.rmdir()Output
Watching mock_watch_dir — initial files: ['demo.txt']
Poll 1: no changes
Poll 2: no changes
Poll 3: new files -> ['late_file.log']
Poll 4: no changes
Final tracked files: ['demo.txt', 'late_file.log']How it works
This script uses the standard library's Path and time modules to create a simple polling watcher. It snapshots the initial set of files, then every poll interval compares the current set to detect new arrivals. The threading demo simulates a late-arriving file to show how new files are caught on a later poll. This mimics the directory-watch behavior often needed in ETL pipelines before rolling into Spark batch jobs.
Common mistakes
- Using `os.listdir` without filtering directories, which can pick up subfolders as files.
- Assuming all new files appear during the first poll; real systems may deliver files with delay.
- Forgetting to clean up test directories and files, leaving stray artifacts.
Variations
- Use `watchdog` for event-driven monitoring instead of polling to avoid latency.
- Track file modification times or hashes to detect overwritten files, not just new names.
Real-world use cases
- Ingesting CSV files dropped into an S3-like staging area before loading into Spark.
- Watching a shared network folder for raw logs to feed into a stream-processing pipeline.
- Automating ETL triggers when partner systems drop data files into a landing zone.
Sponsored
More from Big data & Spark
Keep learning
Related tutorials and quizzes for this topic.