lilith-file-watcher-daemon

File watcher daemon with debouncing and pattern matching, extending lilith-daemon-core.

Features

• Extends IntervalDaemon: Built on top of lilith-daemon-core's robust lifecycle management
• Efficient File Watching: Uses watchfiles library for high-performance file monitoring
• Debouncing: Batches rapid file changes to avoid overwhelming callbacks
• Pattern Matching: Filter files using glob patterns (e.g., **/*.py, *.json)
• Multiple Watch Paths: Monitor multiple directories simultaneously
• Change Type Tracking: Distinguish between added, modified, and deleted files
• Async Callbacks: Fully async/await based for clean integration
• Enable/Disable: Pause and resume file watching without stopping the daemon

Installation

pip install lilith-file-watcher-daemon

Quick Start

import asyncio
from pathlib import Path
from lilith_file_watcher_daemon import FileWatcherDaemon, FileChange

async def on_changes(changes: list[FileChange]):
    """Called when files change (after debounce period)."""
    for change in changes:
        print(f"{change.change_type.value}: {change.path}")

async def main():
    # Create daemon watching current directory for Python files
    daemon = FileWatcherDaemon(
        watch_paths=[Path.cwd()],
        patterns=["**/*.py"],
        callback=on_changes,
        debounce_ms=500,
    )

    # Start watching (runs until stopped)
    await daemon.start()

if __name__ == "__main__":
    asyncio.run(main())

Usage

Basic File Watching

Watch a directory for all file changes:

from pathlib import Path
from lilith_file_watcher_daemon import FileWatcherDaemon, FileChange

async def handle_changes(changes: list[FileChange]):
    print(f"Detected {len(changes)} changes")
    for change in changes:
        print(f"  {change.change_type.value}: {change.path}")

daemon = FileWatcherDaemon(
    watch_paths=[Path("/path/to/watch")],
    callback=handle_changes,
    debounce_ms=1000,  # Wait 1 second after last change
)

await daemon.start()

Pattern Filtering

Watch only specific file types:

daemon = FileWatcherDaemon(
    watch_paths=[Path.cwd()],
    patterns=[
        "**/*.py",      # All Python files
        "**/*.json",    # All JSON files
        "**/config/*.toml",  # TOML files in config directories
    ],
    callback=handle_changes,
)

Multiple Watch Paths

Monitor multiple directories:

daemon = FileWatcherDaemon(
    watch_paths=[
        Path("/path/to/source"),
        Path("/path/to/config"),
        Path("/path/to/data"),
    ],
    callback=handle_changes,
)

Enable/Disable Watching

Pause and resume without stopping the daemon:

# Start watching
await daemon.start()

# Temporarily pause watching
daemon.disable()
print("Watching paused")

# Resume watching
daemon.enable()
print("Watching resumed")

# Stop completely
await daemon.stop()

Lifecycle Management

Control daemon lifecycle:

# Start daemon in background
daemon_task = asyncio.create_task(daemon.start())

# Check state
print(f"Running: {daemon.is_running}")
print(f"Enabled: {daemon.is_enabled}")
print(f"Total cycles: {daemon.total_cycles}")
print(f"Uptime: {daemon.uptime_seconds}s")

# Graceful shutdown
await daemon.stop()
await daemon_task

API Reference

FileWatcherDaemon

Main daemon class for watching files.

class FileWatcherDaemon(IntervalDaemon):
    def __init__(
        self,
        watch_paths: list[Path],
        callback: ChangeCallback,
        patterns: list[str] | None = None,
        debounce_ms: int = 500,
        daemon_id: str | None = None,
    )

Parameters:

• watch_paths: List of directory paths to watch
• callback: Async function called with list of FileChange objects
• patterns: Optional glob patterns to filter files (default: match all)
• debounce_ms: Milliseconds to wait before triggering callback (default: 500)
• daemon_id: Optional unique identifier for this daemon instance

Methods:

• start(): Start the daemon (blocking, runs until stopped)
• stop(): Stop the daemon gracefully
• enable(): Resume file watching
• disable(): Pause file watching

Properties:

• is_running: Whether daemon loop is running
• is_enabled: Whether file watching is enabled
• total_cycles: Number of cycles executed
• uptime_seconds: Daemon uptime in seconds

FileChange

Represents a file system change event.

class FileChange:
    path: Path           # Path to changed file
    change_type: ChangeType  # Type of change (ADDED, MODIFIED, DELETED)

ChangeType

Enum representing type of file change.

class ChangeType(str, Enum):
    ADDED = "added"
    MODIFIED = "modified"
    DELETED = "deleted"

ChangeCallback

Type alias for change callback function.

ChangeCallback: TypeAlias = Callable[[list[FileChange]], Awaitable[None]]

Integration Examples

With Configuration Reloading

Automatically reload configuration when files change:

import json
from pathlib import Path
from lilith_file_watcher_daemon import FileWatcherDaemon, FileChange

class ConfigManager:
    def __init__(self, config_dir: Path):
        self.config_dir = config_dir
        self.config = {}

        self.daemon = FileWatcherDaemon(
            watch_paths=[config_dir],
            patterns=["**/*.json"],
            callback=self.on_config_change,
            debounce_ms=1000,
        )

    async def on_config_change(self, changes: list[FileChange]):
        """Reload configuration when files change."""
        for change in changes:
            if change.change_type == ChangeType.DELETED:
                self.config.pop(change.path.name, None)
            else:
                with change.path.open() as f:
                    self.config[change.path.name] = json.load(f)

        print(f"Configuration reloaded: {len(self.config)} files")

    async def start(self):
        """Start watching for configuration changes."""
        await self.daemon.start()

With Hot Module Reloading

Watch Python files and trigger reloads:

import importlib
import sys
from pathlib import Path
from lilith_file_watcher_daemon import FileWatcherDaemon, FileChange, ChangeType

class HotReloader:
    def __init__(self, watch_dir: Path, module_name: str):
        self.watch_dir = watch_dir
        self.module_name = module_name

        self.daemon = FileWatcherDaemon(
            watch_paths=[watch_dir],
            patterns=["**/*.py"],
            callback=self.on_source_change,
            debounce_ms=500,
        )

    async def on_source_change(self, changes: list[FileChange]):
        """Reload module when source files change."""
        if any(c.change_type != ChangeType.DELETED for c in changes):
            print(f"Source files changed, reloading {self.module_name}...")

            # Reload module
            if self.module_name in sys.modules:
                importlib.reload(sys.modules[self.module_name])
                print("Module reloaded successfully")

    async def start(self):
        await self.daemon.start()

With Build System Trigger

Trigger builds when source files change:

import subprocess
from pathlib import Path
from lilith_file_watcher_daemon import FileWatcherDaemon, FileChange

async def trigger_build(changes: list[FileChange]):
    """Run build command when files change."""
    print(f"{len(changes)} files changed, triggering build...")

    result = subprocess.run(
        ["pnpm", "build"],
        capture_output=True,
        text=True,
    )

    if result.returncode == 0:
        print("Build succeeded")
    else:
        print(f"Build failed: {result.stderr}")

# Watch source directory
daemon = FileWatcherDaemon(
    watch_paths=[Path("src")],
    patterns=["**/*.ts", "**/*.tsx"],
    callback=trigger_build,
    debounce_ms=2000,  # Wait 2 seconds for file activity to settle
)

await daemon.start()

Advanced Usage

Custom Debouncing

Fine-tune debouncing for your use case:

# Short debounce for interactive development
daemon = FileWatcherDaemon(
    watch_paths=[Path("src")],
    callback=handle_changes,
    debounce_ms=200,  # Very responsive
)

# Long debounce for expensive operations
daemon = FileWatcherDaemon(
    watch_paths=[Path("data")],
    callback=rebuild_index,
    debounce_ms=5000,  # Wait 5 seconds for batch operations
)

Pattern Matching Details

Pattern matching supports full glob syntax:

patterns = [
    "*.py",              # Files in watch directory root
    "**/*.py",           # All Python files recursively
    "src/**/*.py",       # Python files under src/
    "test_*.py",         # Test files starting with test_
    "*.{py,pyi}",        # Python source and stub files
    "config/prod/*.json", # JSON in specific directory
]

Error Handling

The daemon inherits error handling from IntervalDaemon:

from lilith_file_watcher_daemon import FileWatcherDaemon

class CustomWatcher(FileWatcherDaemon):
    async def on_error(self, error: Exception) -> bool:
        """Custom error handling."""
        logger.error(f"Watcher error: {error}")

        # Return True to continue, False to stop
        return True  # Keep running despite errors

Testing

Run tests:

pytest tests/

Run tests with coverage:

pytest --cov=lilith_file_watcher_daemon tests/

Dependencies

• lilith-daemon-core>=1.0.0 - Base daemon infrastructure
• watchfiles>=0.21.0 - High-performance file watching

License

MIT License

Related Packages

• lilith-daemon-core - Base daemon infrastructure
• watchfiles - File watching library