pynotify-mcriley821 0.0.2

PyNotify is an async Python interface to the Linux inotify API.

PyNotify is an async Python interface to the Linux inotify API.

See man inotify for more information regarding inotify details.

See the documentation!

Install

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 .

Description

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.

Usage

Simple use case

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

A slightly more interesting example

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.

Adding/Modifying/Removing watches

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

Adding/Removing/Clearing EventHandlers

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)

FAQ

To be filled as questions arise…

License

The UNLICENSE. See https://www.unlicense.org for more info.