asyncinotify

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())