### Registering Watches with Watcher.watch() Source: https://context7.com/rbarrois/aionotify/llms.txt Register directories or files for monitoring with specified event flags. Watches can be added before or after setup. An optional alias can be provided for easier identification of watch sources in events. ```python import aionotify watcher = aionotify.Watcher() # Watch with default alias (uses path as alias) watcher.watch('/var/log', flags=aionotify.Flags.MODIFY) # Watch with custom alias watcher.watch( path='/home/user/documents', flags=aionotify.Flags.CREATE | aionotify.Flags.DELETE, alias='user_docs' ) # Watch for multiple event types watcher.watch( path='/tmp/uploads', flags=aionotify.Flags.CREATE | aionotify.Flags.MODIFY | aionotify.Flags.DELETE | aionotify.Flags.MOVED_TO, alias='uploads' ) ``` -------------------------------- ### Complete File Monitor with Signal Handling Source: https://context7.com/rbarrois/aionotify/llms.txt A comprehensive example of a production-ready file monitoring application. It includes setting up watches, processing events in a loop, handling timeouts, and implementing graceful shutdown using signal handlers. ```python import asyncio import signal import aionotify class FileMonitor: def __init__(self): self.watcher = aionotify.Watcher() self.running = True def add_watch(self, path, alias=None): """Add a path to monitor for common file events.""" flags = ( aionotify.Flags.CREATE | aionotify.Flags.MODIFY | aionotify.Flags.DELETE | aionotify.Flags.MOVED_FROM | aionotify.Flags.MOVED_TO ) self.watcher.watch(path, flags, alias=alias) async def run(self): """Main event processing loop.""" await self.watcher.setup() print("File monitor started. Press Ctrl+C to stop.") while self.running: try: event = await asyncio.wait_for( self.watcher.get_event(), timeout=1.0 ) if event is None: break self.handle_event(event) except asyncio.TimeoutError: continue # Check running flag periodically print("Shutting down...") self.watcher.close() def handle_event(self, event): """Process a single file event.""" flags = aionotify.Flags.parse(event.flags) flag_names = [f.name for f in flags] print(f"[{event.alias}] {event.name}: {', '.join(flag_names)}") def stop(self): """Signal the monitor to stop.""" self.running = False async def main(): monitor = FileMonitor() monitor.add_watch('/tmp', alias='temp') monitor.add_watch('/var/log', alias='logs') # Set up signal handlers for graceful shutdown loop = asyncio.get_running_loop() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, monitor.stop) await monitor.run() if __name__ == '__main__': asyncio.run(main()) ``` -------------------------------- ### Monitor Directory with aionotify Source: https://context7.com/rbarrois/aionotify/llms.txt Set up a watcher to monitor a directory for file modifications and creations. Ensure the watcher is properly closed after use. This example demonstrates the core workflow of setting up, running, and closing a watcher. ```python import asyncio import aionotify async def monitor_directory(): # Create a watcher instance watcher = aionotify.Watcher() # Register paths to watch before or after setup watcher.watch( path='/var/log', flags=aionotify.Flags.MODIFY | aionotify.Flags.CREATE, alias='logs' # Optional alias, defaults to path ) # Initialize the watcher (connects to inotify) await watcher.setup() try: # Continuously get events while True: event = await watcher.get_event() if event is None: break # Watcher was closed print(f"Event: {event.name} in {event.alias}") print(f"Flags: {aionotify.Flags.parse(event.flags)}") finally: watcher.close() asyncio.run(monitor_directory()) ``` -------------------------------- ### Asynchronously Get File System Events Source: https://context7.com/rbarrois/aionotify/llms.txt The get_event() coroutine asynchronously waits for and returns the next file system event. It returns an Event namedtuple with detailed attributes or None if the watcher is closed. This example processes a fixed number of events. ```python import asyncio import aionotify async def process_events(): watcher = aionotify.Watcher() watcher.watch('/tmp/watch_dir', aionotify.Flags.CREATE | aionotify.Flags.MODIFY) await watcher.setup() # Process exactly 5 events for i in range(5): event = await watcher.get_event() if event is None: print("Watcher closed unexpectedly") break # Event attributes: # - event.name: filename that triggered the event # - event.flags: integer bitmask of event flags # - event.alias: alias of the watch that triggered this event # - event.cookie: links MOVED_FROM and MOVED_TO events for renames print(f"Event {i+1}: File '{event.name}' - Flags: {event.flags}") watcher.close() asyncio.run(process_events()) ``` -------------------------------- ### Basic aionotify Usage Source: https://github.com/rbarrois/aionotify/blob/master/README.rst Demonstrates setting up a watcher, adding a watch with an alias and flags, and asynchronously retrieving events. Ensure the watcher is closed after use. ```python import asyncio import aionotify # Setup the watcher watcher = aionotify.Watcher() watcher.watch(alias='logs', path='/var/log', flags=aionotify.Flags.MODIFY) async def work(): await watcher.setup() for _i in range(10): # Pick the 10 first events event = await watcher.get_event() print(event) watcher.close() asyncio.run(work()) ``` -------------------------------- ### Handle file rename events Source: https://context7.com/rbarrois/aionotify/llms.txt Demonstrates how to handle file rename operations by correlating MOVED_FROM and MOVED_TO events using the cookie attribute. Requires setting up a watcher and processing two consecutive events. ```python import asyncio import aionotify async def handle_rename(): watcher = aionotify.Watcher() watcher.watch('/tmp/test', aionotify.Flags.MOVED_FROM | aionotify.Flags.MOVED_TO) await watcher.setup() # When a file is renamed, you get two events with the same cookie event1 = await watcher.get_event() # MOVED_FROM event event2 = await watcher.get_event() # MOVED_TO event # Event structure print(f"Event 1:") print(f" name: {event1.name}") # Original filename print(f" flags: {event1.flags}") # MOVED_FROM flag value print(f" alias: {event1.alias}") # Watch alias print(f" cookie: {event1.cookie}") # Links the two events print(f"Event 2:") print(f" name: {event2.name}") # New filename print(f" flags: {event2.flags}") # MOVED_TO flag value print(f" cookie: {event2.cookie}") # Same as event1.cookie # Verify it's the same rename operation assert event1.cookie == event2.cookie print(f"File renamed from '{event1.name}' to '{event2.name}'") watcher.close() asyncio.run(handle_rename()) ``` -------------------------------- ### Watcher.watch() Method Source: https://context7.com/rbarrois/aionotify/llms.txt Details on how to register paths for monitoring using the watch() method, including options for aliases and multiple event flags. ```APIDOC ## Watcher.watch() The watch() method registers a directory or file path for monitoring with specified event flags. Watches can be added before or after calling setup(). An optional alias provides a friendly name for identifying the watch source in events. ### Method ```python import aionotify watcher = aionotify.Watcher() # Watch with default alias (uses path as alias) watcher.watch('/var/log', flags=aionotify.Flags.MODIFY) # Watch with custom alias watcher.watch( path='/home/user/documents', flags=aionotify.Flags.CREATE | aionotify.Flags.DELETE, alias='user_docs' ) # Watch for multiple event types watcher.watch( path='/tmp/uploads', flags=aionotify.Flags.CREATE | aionotify.Flags.MODIFY | aionotify.Flags.DELETE | aionotify.Flags.MOVED_TO, alias='uploads' ) ``` ``` -------------------------------- ### Watching a Directory without an Alias Source: https://github.com/rbarrois/aionotify/blob/master/README.rst Shows how to add a watch to a directory using its path as the default alias. This is equivalent to explicitly providing the path as the alias. ```python watcher = aionotify.Watcher() watcher.watch('/var/log', flags=aionotify.Flags.MODIFY) ``` -------------------------------- ### Watcher Class Usage Source: https://context7.com/rbarrois/aionotify/llms.txt Demonstrates the basic usage of the Watcher class to monitor a directory for file modifications and creations. ```APIDOC ## Watcher Class The Watcher class is the main interface for setting up and managing file system watches. It handles registration of watch rules, connection to the inotify subsystem, and asynchronous event retrieval. ### Method ```python import asyncio import aionotify async def monitor_directory(): # Create a watcher instance watcher = aionotify.Watcher() # Register paths to watch before or after setup watcher.watch( path='/var/log', flags=aionotify.Flags.MODIFY | aionotify.Flags.CREATE, alias='logs' # Optional alias, defaults to path ) # Initialize the watcher (connects to inotify) await watcher.setup() try: # Continuously get events while True: event = await watcher.get_event() if event is None: break # Watcher was closed print(f"Event: {event.name} in {event.alias}") print(f"Flags: {aionotify.Flags.parse(event.flags)}") finally: watcher.close() asyncio.run(monitor_directory()) ``` ``` -------------------------------- ### Combine flags with bitwise OR Source: https://context7.com/rbarrois/aionotify/llms.txt Combine multiple event flags using the bitwise OR operator to specify desired event types for monitoring. ```python combined = flags.CREATE | flags.MODIFY | flags.DELETE ``` -------------------------------- ### Removing a Watch by Alias Source: https://github.com/rbarrois/aionotify/blob/master/README.rst Illustrates how to remove a previously set watch using its alias. This is useful for dynamically managing monitored directories. ```python watcher = aionotify.Watcher() watcher.watch('/var/log', flags=aionotify.Flags.MODIFY) watcher.unwatch('/var/log') ``` -------------------------------- ### Watcher.get_event() Coroutine Source: https://context7.com/rbarrois/aionotify/llms.txt Describes the get_event() coroutine for asynchronously retrieving file system events and the structure of the returned Event object. ```APIDOC ## Watcher.get_event() The get_event() coroutine asynchronously waits for and returns the next file system event. Returns an Event namedtuple with flags, cookie, name, and alias attributes. Returns None if the watcher is closed. ### Method ```python import asyncio import aionotify async def process_events(): watcher = aionotify.Watcher() watcher.watch('/tmp/watch_dir', aionotify.Flags.CREATE | aionotify.Flags.MODIFY) await watcher.setup() # Process exactly 5 events for i in range(5): event = await watcher.get_event() if event is None: print("Watcher closed unexpectedly") break # Event attributes: # - event.name: filename that triggered the event # - event.flags: integer bitmask of event flags # - event.alias: alias of the watch that triggered this event # - event.cookie: links MOVED_FROM and MOVED_TO events for renames print(f"Event {i+1}: File '{event.name}' - Flags: {event.flags}") watcher.close() asyncio.run(process_events()) ``` ``` -------------------------------- ### Flags Enum Source: https://context7.com/rbarrois/aionotify/llms.txt Overview of the Flags enum, which defines inotify event types and watch modifiers, and how to use the parse() method. ```APIDOC ## Flags Enum The Flags enum defines all inotify event types and watch modifiers. Use bitwise OR to combine multiple flags. The parse() class method converts a flag bitmask back into a list of individual flag values. ### Method ```python import aionotify # Common event flags flags = aionotify.Flags # File content events flags.ACCESS # File was accessed (read) flags.MODIFY # File was modified (write) flags.OPEN # File was opened flags.CLOSE_WRITE # Writable file was closed flags.CLOSE_NOWRITE # Read-only file was closed # File metadata events flags.ATTRIB # Metadata changed (permissions, timestamps, etc.) ``` ``` -------------------------------- ### Watcher.unwatch() Method Source: https://context7.com/rbarrois/aionotify/llms.txt Explains how to remove a registered watch using its alias with the unwatch() method. ```APIDOC ## Watcher.unwatch() The unwatch() method removes a previously registered watch using its alias. This stops monitoring the associated path and cleans up internal resources. ### Method ```python import asyncio import aionotify async def selective_monitoring(): watcher = aionotify.Watcher() # Set up multiple watches watcher.watch('/var/log', aionotify.Flags.MODIFY, alias='logs') watcher.watch('/tmp', aionotify.Flags.CREATE, alias='temp') await watcher.setup() # Process some events... event = await watcher.get_event() print(f"Got event: {event}") # Remove the temp watch (using alias) watcher.unwatch('temp') # Continue monitoring only /var/log # Events from /tmp will no longer be received watcher.close() asyncio.run(selective_monitoring()) ``` ``` -------------------------------- ### Remove Watch with Watcher.unwatch() Source: https://context7.com/rbarrois/aionotify/llms.txt Remove a previously registered watch using its alias. This stops monitoring the associated path and cleans up resources. This is useful for dynamically changing which directories are being monitored. ```python import asyncio import aionotify async def selective_monitoring(): watcher = aionotify.Watcher() # Set up multiple watches watcher.watch('/var/log', aionotify.Flags.MODIFY, alias='logs') watcher.watch('/tmp', aionotify.Flags.CREATE, alias='temp') await watcher.setup() # Process some events... event = await watcher.get_event() print(f"Got event: {event}") # Remove the temp watch (using alias) watcher.unwatch('temp') # Continue monitoring only /var/log # Events from /tmp will no longer be received watcher.close() asyncio.run(selective_monitoring()) ``` -------------------------------- ### Parse event flags Source: https://context7.com/rbarrois/aionotify/llms.txt Parse an integer bitmask representing event flags into a list of human-readable flag names. This is useful for interpreting raw event data. ```python event_flags = 0x00000102 parsed = aionotify.Flags.parse(event_flags) print(parsed) ``` -------------------------------- ### aionotify Flags Enum Source: https://context7.com/rbarrois/aionotify/llms.txt Defines inotify event types and watch modifiers. Use bitwise OR to combine flags. The parse() class method converts a flag bitmask back into a list of individual flag values. Common event flags are available directly. ```python import aionotify # Common event flags flags = aionotify.Flags # File content events flags.ACCESS # File was accessed (read) flags.MODIFY # File was modified (write) flags.OPEN # File was opened flags.CLOSE_WRITE # Writable file was closed flags.CLOSE_NOWRITE # Read-only file was closed # File metadata events flags.ATTRIB # Metadata changed (permissions, timestamps, etc.) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.