asyncinotify

4.4.4 · active · verified Wed Apr 15

asyncinotify is a simple, optionally-async Python inotify library, focusing on ease of use and modern Python features. It provides a Pythonic interface to Linux's inotify API, built on `ctypes` without external dependencies. The library is actively maintained, currently at version 4.4.4, and supports Python 3.6 and newer. It offers both asynchronous (async/await) and synchronous modes of operation for file system event monitoring.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to asynchronously watch a temporary directory for common file system events (create, modify, access, delete). It uses `Inotify` as a context manager to ensure the inotify handle is properly closed. Events are iterated asynchronously, printing details as they occur.

import asyncio
import os
import tempfile
from pathlib import Path
from asyncinotify import Inotify, Mask

async def main():
    # Create a temporary directory for watching
    with tempfile.TemporaryDirectory() as tmpdir_name:
        watch_path = Path(tmpdir_name)
        print(f"Watching directory: {watch_path}")

        with Inotify() as inotify:
            # Add a watch for various events
            inotify.add_watch(
                watch_path,
                Mask.ACCESS | Mask.MODIFY | Mask.OPEN | Mask.CREATE |
                Mask.DELETE | Mask.ATTRIB | Mask.CLOSE | Mask.MOVE | Mask.ONLYDIR
            )
            print("Watch added. Creating a file in the directory...")

            # Create a file to trigger events
            test_file = watch_path / "test_file.txt"
            test_file.write_text("hello")
            await asyncio.sleep(0.1) # Give a moment for events to register
            test_file.unlink()
            await asyncio.sleep(0.1) # Give a moment for events to register
            print("File operations completed. Collecting events...")

            # Collect a few events. In a real application, you'd process them in a loop.
            events_collected = 0
            async for event in inotify:
                print(f"Event: {event}")
                events_collected += 1
                # Break after a few events or based on application logic
                if events_collected >= 3: # Adjust based on expected events
                    break
            print("Finished collecting events.")

asyncio.run(main())

view raw JSON →