Trigger a Pipeline When a New File Appears in a Directory

Poll a directory every 0.5 seconds and return the name of the first new file that appears, or None after a timeout.

Easy Python 3.10+ Aug 9, 2026 Data pipelines & processing 13 views 0 copies

polling filesystem file-watcher

Python code

31 lines

Python 3.10+

import time
from pathlib import Path


def watch_for_file(directory: str, interval: float = 0.5, timeout: float = 10.0) -> str | None:
    """Poll a directory and trigger when a new file appears."""
    watch_dir = Path(directory)
    watch_dir.mkdir(exist_ok=True)
    
    known_files = set(watch_dir.iterdir())
    start = time.time()
    
    while time.time() - start < timeout:
        current_files = set(watch_dir.iterdir())
        new_files = current_files - known_files
        if new_files:
            return new_files.pop().name
        time.sleep(interval)
    return None


if __name__ == "__main__":
    import tempfile
    import os
    
    with tempfile.TemporaryDirectory() as tmpdir:
        result = watch_for_file(tmpdir, interval=0.1, timeout=1.0)
        if result:
            print(f"New file detected: {result}")
        else:
            print("No new files detected")

Output

stdout

No new files detected

How it works

The function builds a set of existing filenames before the loop, then repeatedly snapshots the directory and computes the set difference to find newly added files. Using set difference is O(n) and avoids scanning file metadata, making the poll cheap. The time.sleep call throttles CPU usage, and the timeout guard prevents an infinite loop. Returning the first new file name (via pop()) is non-deterministic when multiple files appear, but the caller can expand this to handle a batch. The mkdir(exist_ok=True) ensures the watched directory exists, preventing errors on the first poll.

Common mistakes

Forgetting to call `mkdir(exist_ok=True)` so the watch fails if the directory doesn't exist.
Using `glob` instead of `iterdir()`, which can miss hidden files and is slower.
Not converting the previous snapshot to a set, causing O(n²) comparisons.
Returning immediately on the first diff instead of collecting a batch, which may miss files added milliseconds after the snapshot.

Variations

Use `watchdog` library for event-driven monitoring instead of polling.
Add a callback parameter so the function runs a pipeline step on detection.

Real-world use cases

Watching an SFTP drop folder for a new data export and triggering an ingestion job.
Monitoring a shared network drive for a client upload and starting a validation script.
Triggering a downstream ETL step when a teammate drops a CSV into a synced folder.

Trigger a Pipeline When a New File Appears in a Directory

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Data pipelines & processing

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Data pipelines & processing

Keep learning

Tutorials

Quizzes