PyNotify is an async Python interface to the Linux inotify API.
See man inotify for more information regarding inotify details.
See the documentation!
Install via pip:
pip install pynotify-mcriley821
To install manually, clone the repo and pip install:
git clone https://github.com/mcriley821/PyNotify.git cd PyNotify && pip install .
PyNotify uses the ctypes module to interface with the inotify API to allow the user to create 'watches' for monitoring filesystem events. These events are parsed into Event objects, which are then handled by EventHandler objects.
Any number of EventHandlers can be added to a Notifier instance to handle a
specific watch. This is done when requesting a watch via Notifier.add_watch
When an Event is emitted for the corresponding watch, each EventHandler is
queried for capability of handling said Event. The Event is subsequently
passed to the EventHandler if it is capable.
As an example, an EventHandler that handles all event types for a watch could be defined as so:
class AllHandler: def handle_event(self, event: Event) -> None: # just print out what is happening print(f"{event.type.name} at {event.file_path}") def can_handle_event_type(self, type: EventType) -> bool: return EventType.ALL & type != 0
The AllHandler can now be added to a watch via
Notifier.add_watch:
async def main(): with pynotify.Notifier() as notifier: notifier.add_watch(pathlib.Path.cwd(), AllHandler()) await notifier.run()
class OpenHandler: def handle_event(self, event: Event) -> None: ... def can_handle_event_type(self, type: EventType) -> bool: return EventType.OPEN & type != 0 class CloseHandler: def handle_event(self, event: Event) -> None: ... def can_handle_event_type(self, type: EventType) -> bool: return EventType.CLOSE & type != 0 async def stop_loop(stop_event: asyncio.Event): await asyncio.sleep(10) stop_event.set() async def main(): with pynotify.Notifier() as notifier: path = pathlib.Path.cwd() stop_event = asyncio.Event() notifier.add_watch(path, OpenHandler(), CloseHandler(), only_event_types=EventType.OPEN | EventType.CLOSE) await asyncio.gather( notifier.run(stop_event=stop_event), stop_loop(stop_event))
The above example will run the Notifier run-loop for 10 seconds, generating only open and close Events for the watch on the current working directory.
Watches can be added as simply as we've seen above. There are a few more options that can be specified when adding a watch:
async def main(): with pynotify.Notifier() as notifier: path = pathlib.Path.cwd() notifier.add_watch( path, # path to add a watch on # any number of handlers for the watch AllHandler(), OpenHandler(), CloseHandler(), # restrict EventTypes generated by the watch only_event_types=EventTypes.OPEN, # raises if False and path is a symlink follow_symlinks=False, # raises if True and path is not a directory if_directory_only=True, # if True, generate a single event then remove the watch oneshot=False, # See the docs for more info on this flag exclude_unlinks=True)
EventTypes for a watch can be modified after it has been added to a Notifier:
async def main(): with pynotify.Notifier() as notifier: path = pathlib.Path.cwd() notifier.add_watch(path) # generates all EventTypes by default ... # generate only CLOSE Events notifier.modify_watch_event_type(path, EventType.CLOSE) # merge EventTypes to generate both CLOSE and OPEN Events notifier.modify_watch_event_type(path, EventType.OPEN, merge=True)
Watches are easily removed:
async def main(): with pynotify.Notifier() as notifier: path = pathlib.Path.cwd() notifier.add_watch(path) ... notifier.remove_watch(path) # notifier.remove_watch(path) # raises, since path not being watched notifier.remove_watch(path, raises=False) # don't raise
EventHandlers can be added when adding a watch, and can be added or removed after a watch has already been established:
async def main(): with pynotify.Notifier() as notifier: path = pathlib.Path.cwd() open_handler = OpenHandler() notifier.add_watch(path, open_handler) # add open_handler to watch all_handler = AllHandler() # add all_handler and a CloseHandler notifier.add_handlers(path, all_handler, CloseHandler()) # remove only the all_handler notifier.remove_handlers(path, all_handler) # clear all handlers on the watch notifier.clear_handlers(path)
Note in the above example that the Notifier.add_watches and
Notifier.remove_handlers method can take any number of EventHandlers
to add or remove. Also, duplicate handlers for a watch are not possible, and
removing a handler that isn't on a watch will do nothing:
async def main(): with pynotify.Notifier() as notifier: path = pathlib.Path.cwd() open_handler = OpenHandler() notifier.add_watch(path, open_handler) # does nothing, since open_handler already on the watch! notifier.add_handlers(path, open_handlers) notifier.remove_handlers(path, open_handler) # no more handlers # does nothing, since open_handler isn't on the watch notifier.remove_handlers(path, open_handler)
To be filled as questions arise...
The UNLICENSE. See https://www.unlicense.org for more info.