PyNotify is an async Python interface to the Linux inotify API.
See man inotify <https://man7.org/linux/man-pages/man7/inotify.7.html>
_
for more information regarding inotify details.
See the documentation <https://mcriley821.github.io/PyNotify/build/html/index.html>
_!
To install manually, clone the repo and pip install:
.. code:: bash
git clone https://github.com/mcriley821/PyNotify.git cd PyNotify && pip install .
PyNotify uses the ctypes <https://docs.python.org/3/library/ctypes.html>
_
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:
.. role:: python(code) :language: python
.. code:: python
class AllHandler: def handle_event(self, event: Event) -> None:
print(f"{event.type.name} at {event.file_path}")
def can_handle_event_type(self, type: EventType) -> bool:
return EventType.ALL & type != 0
The :python:AllHandler
can now be added to a watch via
:python:Notifier.add_watch
:
.. code:: python
async def main(): with pynotify.Notifier() as notifier: notifier.add_watch(pathlib.Path.cwd(), AllHandler()) await notifier.run()
.. code:: python
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:
.. code:: python
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:
.. code:: python
async def main(): with pynotify.Notifier() as notifier: path = pathlib.Path.cwd() notifier.add_watch(path) # generates all EventTypes by default ...
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:
.. code:: python
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=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:
.. code:: python
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 :python:Notifier.add_watches
and
:python: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:
.. code:: python
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.