Real-Time File Monitoring with Python

This article presents a robust solution for monitoring directory changes in real-time using Python's inotify interface combined with multiprocessing to handle file events asynchronously.

Overview of dirwatcher

The dirwatcher project provides a lightweight file system watcher that monitors a specified directory for changes and automatically triggers actions when file operations occur. Whether you need to process uploaded files, react to file deletions, or handle attribute changes, this solution offers a clean, efficient approach.

GitHub Repository: https://github.com/MourIdri/dirwatcher

Why Monitor Directories?

Directory monitoring is essential for:

• File Processing Pipelines: Automatically process files as they're uploaded or moved to a directory
• Real-time Analytics: React immediately to new data files in monitoring systems
• Log Processing: Track and process new log files as they're generated
• Backup Automation: Trigger backups when files are modified
• Upload Management: Handle file uploads without polling databases
• Event-Driven Workflows: Build reactive systems that respond to file system changes

Understanding inotify

inotify is a Linux kernel subsystem that provides efficient file system event notification. Unlike polling-based approaches that continuously check directories, inotify delivers events immediately when changes occur.

Key Advantages of inotify

• Efficiency: No polling overhead; kernel notifies application of changes
• Real-time: Instant notifications of file system events
• Selective: Monitor only specific directories and event types
• Scalable: Handles monitoring many files without performance degradation

Important inotify Events

The dirwatcher monitors these key events:

Installation and Setup

Prerequisites

pip install inotify_simple
# or
pip install inotify-adapters

Basic Configuration

Before running the watcher, configure these settings in wa_dir.py:

# Path to monitor (change to your target directory)
path = "/mnt/c/code_source/GENERAL/HPC/PARA/PATHTEST/"

# Log file name (customize as needed)
fileLogName = 'LOG_directory_operations.log'

# Rotating log handler configuration
maxBytes = 1048576 * 5  # 5MB
backupCount = 7  # Keep 7 backup log files

Architecture and Implementation

Component 1: Logging Configuration

Proper logging is critical for monitoring applications:

import logging
from logging.handlers import RotatingFileHandler

_DEFAULT_LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
_LOGGER = logging.getLogger(__name__)

def _configure_logging():
    _LOGGER.setLevel(logging.INFO)

    # Console handler
    ch = logging.StreamHandler()
    formatter = logging.Formatter(_DEFAULT_LOG_FORMAT)
    ch.setFormatter(formatter)

    # File handler with rotation
    fileLogName = 'LOG_directory_operations.log'
    fileHandler = logging.handlers.RotatingFileHandler(
        fileLogName,
        maxBytes=(1048576 * 5),  # 5MB
        backupCount=7
    )
    fileHandler.setFormatter(formatter)

    _LOGGER.addHandler(ch)
    _LOGGER.addHandler(fileHandler)

Why Rotating Logs?

• Prevents single log files from consuming unlimited disk space
• Automatically archives old logs
• Maintains historical records while managing storage

Component 2: Event Processing

The core event handler processes file system events:

def PopUpMessage(event):
    filepartname = ".filepart"
    filelaststate_writting = "IN_CLOSE_WRITE"
    filelaststate_deleting = "IN_DELETE"
    filelaststate_in_attrib = "IN_ATTRIB"

    if event is not None:
        (header, type_names, watch_path, filename) = event

        _LOGGER.info(
            "WD=(%d) MASK=(%d) COOKIE=(%d) LEN=(%d) MASK->NAMES=%s "
            "WATCH-PATH=[%s] FILENAME=[%s]",
            header.wd, header.mask, header.cookie, header.len, type_names,
            watch_path.decode('utf-8'), filename.decode('utf-8')
        )

        time.sleep(0.5)  # Small delay to ensure file operation completes

Handling Upload States

The watcher intelligently handles partial file uploads:

# File still uploading (contains .filepart extension)
if filepartname in filename:
    _LOGGER.info(
        'File is still uploading locally - analysis will be done '
        'once the file is completely uploaded'
    )

# File deletion detected
if filelaststate_deleting in type_names and filepartname not in filename:
    _LOGGER.info('File is being deleted from the local repository')

# File attributes added - indicates upload completion
if filelaststate_in_attrib in type_names and filename:
    _LOGGER.info('File completely uploaded locally - attributes added')
    filepushedforwatsonanalysis = str(filename)
    _LOGGER.info(
        'File ready for processing: %s' % filepushedforwatsonanalysis
    )

Component 3: Multiprocessing for Parallel Processing

Multiprocessing enables non-blocking event handling:

def runInParallel(*fns):
    """Execute multiple functions in parallel processes"""
    proc = []
    for fn in fns:
        p = multiprocessing.Process(target=fn)
        p.start()
        proc.append(p)

    # Wait for all processes to complete
    for p in proc:
        p.join()

Why Multiprocessing?

• Non-blocking: Events don't wait for previous processing to complete
• Scalability: Handle multiple file events simultaneously
• Resilience: One processing failure doesn't block other events
• CPU Utilization: Leverage multi-core systems effectively

Component 4: Main Watcher Loop

def My_main(path):
    # Initialize inotify
    i = inotify.adapters.Inotify()
    DirWatcher = i.add_watch(path)

    try:
        # Infinite loop watching for events
        while True:
            # Process each event in the queue
            for event in i.event_gen():
                # Handle event in separate process
                m = multiprocessing.Process(
                    target=PopUpMessage, 
                    args=(event,)
                )
                m.start()
    finally:
        # Cleanup: remove watch
        i.remove_watch(b'/PARA')

if __name__ == '__main__':
    _configure_logging()
    path = "/mnt/c/code_source/GENERAL/HPC/PARA/PATHTEST/"

    # Start watcher in main process
    N = multiprocessing.Process(target=My_main, args=(path,))
    N.start()

Running the Application

Starting the Watcher

python wa_dir.py

The application will:

1. Start monitoring the configured directory
2. Log all file system events to console and file
3. Process events in parallel using multiprocessing
4. Continue running indefinitely until interrupted

Monitoring the Logs

In another terminal, watch the log file in real-time:

tail -f LOG_directory_operations.log

Advanced Configuration

Monitoring Multiple Directories

Extend the watcher to monitor multiple paths:

def watch_multiple_directories():
    paths = [
        "/path/to/uploads/",
        "/path/to/processed/",
        "/path/to/archive/"
    ]

    processes = []
    for path in paths:
        p = multiprocessing.Process(target=My_main, args=(path,))
        p.start()
        processes.append(p)

    # Keep processes running
    for p in processes:
        p.join()

if __name__ == '__main__':
    _configure_logging()
    watch_multiple_directories()

Adding Custom Processing Logic

Replace or extend PopUpMessage to trigger your own logic:

def process_file(filename):
    """Your custom file processing logic"""
    print(f"Processing: {filename}")
    # Add your business logic here
    # Examples: resize images, extract metadata, validate formats, etc.

def PopUpMessage_Extended(event):
    # ... existing code ...
    if filelaststate_in_attrib in type_names and filename:
        filename_str = filename.decode('utf-8')
        _LOGGER.info(f'Processing file: {filename_str}')
        process_file(filename_str)

Filtering Specific File Types

def should_process_file(filename):
    """Filter files by extension"""
    valid_extensions = ('.png', '.jpg', '.pdf', '.txt', '.csv')
    filename_str = filename.decode('utf-8') if isinstance(filename, bytes) else filename
    return any(filename_str.endswith(ext) for ext in valid_extensions)

def PopUpMessage_Filtered(event):
    # ... existing code ...
    if filelaststate_in_attrib in type_names and filename:
        if should_process_file(filename):
            _LOGGER.info(f'Processing valid file: {filename}')
            # Trigger processing
        else:
            _LOGGER.info(f'Skipping invalid file: {filename}')

Best Practices

1. Error Handling

try:
    m = multiprocessing.Process(target=PopUpMessage, args=(event,))
    m.start()
except Exception as e:
    _LOGGER.error(f'Error processing event: {str(e)}')

2. Resource Cleanup

• Always implement try-finally blocks for resource cleanup
• Remove watches when done monitoring
• Close file handlers properly

3. Log Rotation

• Configure appropriate maxBytes and backupCount
• Monitor log directory space usage
• Implement log retention policies

4. Process Management

• Monitor spawned processes for memory leaks
• Implement process pooling for high-volume scenarios
• Add timeouts for stuck processes

5. Path Configuration

• Use absolute paths for reliability
• Verify directory permissions before monitoring
• Handle symbolic links carefully

Performance Considerations

Tuning for High-Volume Scenarios

# For high-frequency events, consider:

# 1. Batch processing
event_queue = []
BATCH_SIZE = 10

def batch_process(queue):
    while len(queue) < BATCH_SIZE:
        # Wait for more events
        pass
    # Process batch together

# 2. Process pool instead of unlimited processes
from multiprocessing import Pool
pool = Pool(processes=4)

Avoiding Common Issues

Common Use Cases

1. Automated Image Processing

def process_image(filename):
    # Resize, optimize, or convert images
    pass

2. File Format Validation

def validate_file_format(filename):
    # Verify file signatures and formats
    pass

3. Upload Virus Scanning

def scan_file(filename):
    # Run virus/malware scans on uploaded files
    pass

4. Metadata Extraction

def extract_metadata(filename):
    # Extract and index file metadata
    pass

5. Automated Backup

def backup_file(filename):
    # Copy to backup location or cloud storage
    pass

Troubleshooting

Issue: inotify watch limit exceeded

# Check current limit
cat /proc/sys/fs/inotify/max_user_watches

# Increase the limit (requires root)
echo 524288 | sudo tee /proc/sys/fs/inotify/max_user_watches

Issue: Process not starting

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)

Issue: Events not captured

• Verify directory path exists
• Check file permissions
• Ensure inotify module is installed
• Verify path is not on network mount

Deployment Considerations

Running as Service

Create a systemd service for automatic startup:

[Unit]
Description=Directory Watcher Service
After=network.target

[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/dirwatcher
ExecStart=/usr/bin/python3 /opt/dirwatcher/wa_dir.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

Enable and start:

sudo systemctl enable dirwatcher
sudo systemctl start dirwatcher

Docker Containerization

FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY wa_dir.py .
CMD ["python", "wa_dir.py"]

Conclusion

The dirwatcher project demonstrates an efficient, production-ready solution for real-time file system monitoring in Python. By combining inotify for event detection with multiprocessing for concurrent handling, you can build responsive file processing pipelines that scale gracefully.

Key Takeaways:

• Use inotify for efficient, non-polling file monitoring
• Leverage multiprocessing for concurrent event handling
• Implement proper logging with rotation for operational visibility
• Handle partial uploads correctly with file extension detection
• Monitor resource usage and implement cleanup routines

GitHub Repository: https://github.com/MourIdri/dirwatcher

File system monitoring is a fundamental capability for building event-driven systems and automated workflows. Use this foundation to build custom solutions for your specific needs.