File Watching

def watch(
    *paths: Union[Path, str],
    watch_filter: Optional[Callable[[Change, str], bool]] = DefaultFilter(),
    debounce: int = 1_600,
    step: int = 50,
    stop_event: Optional[AbstractEvent] = None,
    rust_timeout: int = 5_000,
    yield_on_timeout: bool = False,
    debug: Optional[bool] = None,
    raise_interrupt: bool = True,
    force_polling: Optional[bool] = None,
    poll_delay_ms: int = 300,
    recursive: bool = True,
    ignore_permission_denied: Optional[bool] = None,
) -> Generator[Set[FileChange], None, None]:
    """
    Watch one or more paths and yield a set of changes whenever files change.

    Parameters:
    - *paths: Filesystem paths to watch (files or directories)
    - watch_filter: Callable to filter changes, defaults to DefaultFilter()
    - debounce: Maximum time in ms to group changes before yielding
    - step: Time in ms to wait for new changes if at least one detected
    - stop_event: Event to stop watching (any object with is_set() method)
    - rust_timeout: Maximum time in ms to wait for changes in rust code
    - yield_on_timeout: If True, yield empty set on timeout
    - debug: Print filesystem changes to stdout (uses WATCHFILES_DEBUG env var if None)
    - raise_interrupt: Whether to re-raise KeyboardInterrupt or suppress and stop
    - force_polling: Force use of polling instead of native notifications
    - poll_delay_ms: Delay between polls when force_polling=True
    - recursive: Watch subdirectories recursively
    - ignore_permission_denied: Ignore permission denied errors

    Yields:
    Set[FileChange]: Set of (Change, path) tuples representing file changes

    Raises:
    KeyboardInterrupt: If raise_interrupt=True and Ctrl+C pressed
    """

from watchfiles import watch, Change

# Basic usage - watch current directory
for changes in watch('.'):
    print(f'Changes detected: {changes}')

# Watch multiple paths with custom filter
from watchfiles import PythonFilter

for changes in watch('./src', './tests', watch_filter=PythonFilter()):
    for change_type, path in changes:
        if change_type == Change.added:
            print(f'New file: {path}')
        elif change_type == Change.modified:
            print(f'Modified: {path}')
        elif change_type == Change.deleted:
            print(f'Deleted: {path}')

# Watch with stop event
import threading

stop_event = threading.Event()
threading.Timer(10.0, stop_event.set).start()  # Stop after 10 seconds

for changes in watch('./src', stop_event=stop_event):
    print(f'Changes: {changes}')

async def awatch(
    *paths: Union[Path, str],
    watch_filter: Optional[Callable[[Change, str], bool]] = DefaultFilter(),
    debounce: int = 1_600,
    step: int = 50,
    stop_event: Optional[AnyEvent] = None,
    rust_timeout: Optional[int] = None,
    yield_on_timeout: bool = False,
    debug: Optional[bool] = None,
    raise_interrupt: Optional[bool] = None,
    force_polling: Optional[bool] = None,
    poll_delay_ms: int = 300,
    recursive: bool = True,
    ignore_permission_denied: Optional[bool] = None,
) -> AsyncGenerator[Set[FileChange], None]:
    """
    Asynchronous version of watch() using threads.

    Parameters:
    Same as watch() except:
    - stop_event: anyio.Event, asyncio.Event, or trio.Event
    - rust_timeout: None means 1000ms on Windows, 5000ms elsewhere
    - raise_interrupt: Deprecated, KeyboardInterrupt handled by event loop

    Yields:
    Set[FileChange]: Set of (Change, path) tuples representing file changes
    
    Note:
    KeyboardInterrupt cannot be suppressed and must be caught at the 
    asyncio.run() level or equivalent.
    """

import asyncio
from watchfiles import awatch

async def monitor_files():
    async for changes in awatch('./src', './tests'):
        print(f'Async changes: {changes}')

# Run with proper KeyboardInterrupt handling
try:
    asyncio.run(monitor_files())
except KeyboardInterrupt:
    print('Monitoring stopped')

# With stop event
async def monitor_with_timeout():
    stop_event = asyncio.Event()
    
    # Stop after 10 seconds
    async def stop_after_delay():
        await asyncio.sleep(10)
        stop_event.set()
    
    stop_task = asyncio.create_task(stop_after_delay())
    
    async for changes in awatch('./src', stop_event=stop_event):
        print(f'Changes: {changes}')
    
    await stop_task

asyncio.run(monitor_with_timeout())

# Force polling with custom delay
for changes in watch('./src', force_polling=True, poll_delay_ms=100):
    print(f'Polled changes: {changes}')

# Type definitions used in file watching
FileChange = Tuple[Change, str]

class AbstractEvent(Protocol):
    def is_set(self) -> bool: ...

# Union type for async events (TYPE_CHECKING only)
AnyEvent = Union[anyio.Event, asyncio.Event, trio.Event]