inotify_lite¶

inotify_lite provides a Python 3 wrapper around the Linux inotify API. This lets you monitor filesystem events, and execute callbacks. See inotify(7).

Requires¶

Linux >= 2.6.13 (or glibc >= 2.5)
Python >= 3.6

Installation¶

Install inotify_lite by running:

pip install inotify_lite

Usage¶

To use inotify_lite:

Create an Inotify instance, passing the name of the files (or directories) you wish to watch;
Register a handler (or many), a callable of two arguments:
- an Inotify instance; and
- an InotifyEvent instance.
call Inotify.read to read once, or Inotify.watch to watch until a keyboard interrupt is received.

Examples:

def my_callback(_, event):
    print(event.name)
    print(event.mask)

flags = INFlags.CREATE | INFlags.DELETE
watcher = Inotify("/home/", watch_flags=flags)
watcher.register_handler(INFlags.ALL_FLAGS, my_callback, exclusive=False)
watcher.watch()

def my_callback(_, event):
    print(event.name)
    print(event.mask)

watcher = TreeWatcher("/home/", watch_flags=INFlags.OPEN, timeout=10)
# Watch the home directory for OPEN events with a 10 second timeout.

watcher.register_handler(INFlags.ALL_FLAGS, my_callback, exclusive=False)
watcher.read_once()

The TreeWatcher class is provided to recursively watch directories.

See the documentation for details and options.

Contribute¶

Contributions are welcome. Open an issue for visibility.

To install the dev requirements run python setup.py -e .[dev].

To run the tests run make test.

Code should be formatted with black.

Issue Tracker: https://github.com/jams2/inotify_lite/issues
Source Code: https://github.com/jams2/inotify_lite.git

Support¶

Open an issue.

License¶

The project is licensed under GPLv3.

API¶

class inotify_lite.INFlags(value)¶: Flags defined in <sys/inotify.h>, <bits/inotify.h>.

class inotify_lite.Inotify(*files: str, watch_flags: inotify_lite.inotify.INFlags = <INFlags.NO_FLAGS: 0>, n_buffers: int = 1, buf_size: int = 1024, timeout: Optional[Union[float, int]] = None)¶

Base class for TreeWatcher. Interfaces with inotify(7).

While TreeWatcher provides functionality for watching directories recursively, this is suitable for watching a file (or files).

exclusive_handlers¶

maps INFlags to sets of `Inotify.EventHandler`s. Handler is executed iff event.mask == handler mask.

Type: dict

inclusive_handlers¶

maps INFlags to sets of `Inotify.EventHandler`s. Handler will be executed if a bitwise AND of the event.mask with the handler mask is non-zero.

Type: dict

inotify_fd¶

file descriptor returned by call to inotify_init1.

Type: int

watch_flags¶

flags to be passed to inotify_add_watch.

Type: INFlags

watch_fds¶

a dict mapping watch descriptors to their associated filenames.

Type: dict

timeout¶: time (in seconds) to wait before reading, or None.

files¶

a set of filenames currently being watched.

Type: set

n_buffers¶

number of per instance read buffers.

Type: int

read_buffers¶

per instance read buffers.

Type: list

max_read¶

maximum bytes we can read.

Type: int

buf_size¶

in bytes.

Type: int

LEN_OFFSET¶

we need to read the length of the name before unpacking the bytes to the struct format. See the underlying struct inotify_event.

Type: int

EventHandler¶: alias of typing.Callable

static get_event_struct_format(name_len: int) → str¶

Given an event with name of length (name_len), return the correct format string to pass to struct.unpack.

Parameters

name_len (int) – length of event name, taken from
inotify_event.len. (struct) –

Returns

the struct format string.

read_once() → int¶: Wait self.timeout

register_handler(event_mask: inotify_lite.inotify.INFlags, handler: Callable[[Inotify, InotifyEvent], None], exclusive=True)¶

Register a handler for matching events.

An exclusive handler will only be called if the event mask of the event is an exact match for the specified event_mask.

In contrast, an inclusive handler will be called for any partial match of the event mask. For example, an event returning a mask of IN_ISDIR|IN_OPEN will cause an inclusive handler registered for IN_OPEN to be executed. An exclusive handler registered for IN_OPEN will not be executed in this case.

Parameters

event_mask (INFlags) – event mask to match.
handler (Inotify.EventHandler) – handler to call on matching event.
exclusive (bool) – whether to register it as an exclusive handler (otherwise, it will be inclusive).

watch()¶

Start the read -> callback loop, teardown gracefully.

Python documentation states that select will return an empty list if the process receives a signal before end of timeout or the observed fd is ready. In that case, just break out of the loop. https://docs.python.org/3/library/selectors.html#selectors.BaseSelector.select

class inotify_lite.InotifyEvent(wd: int, mask: int, cookie: int, name_len: int, name: bytes)¶

Maps to struct inotify_event from inotify.h. The from_struct classmethod is provided for convenience.

wd¶

watch descriptor.

Type: int

mask¶

event mask

Type: INFlags

cookie¶

unique id associating IN_MOVED_FROM events with corresponding IN_MOVED_TO.

Type: int

name_len¶

length of name string (len in underlying struct).

Type: int

name¶

name of watched file that event refers to.

Type: string

classmethod from_struct(struct_members: Tuple) → inotify_lite.inotify.InotifyEvent¶

Returns a new InotifyEvent instance from (tuple) result of calling struct.unpack on bytes, with struct_event format.

Parameters

struct_members (tuple) – tuple returned from struct.unpack. Expects members wd (int), mask (int), cookie (int), name_len (int), name (bytes).

Returns

A new InotifyEvent instance.

Raises

ValueError – received a tuple of incorrect length.
TypeError – one of the struct_members members was of incorrect type.

static str_from_bytes(byte_obj: bytes) → str¶

Convert null terminated bytes to Python string.

Parameters: byte_obj (bytes) – bytes representing a null-terminated string.
Returns: a Python string.

class inotify_lite.TreeWatcher(*dirs: str, watch_subdirs: bool = True, watch_flags: inotify_lite.inotify.INFlags = <INFlags.ALL_EVENTS: 4095>, timeout: Optional[Union[float, int]] = None)¶

Watch directories, and optionally all subdirectories.

register_handler(event_mask: inotify_lite.inotify.INFlags, handler: Callable[[Inotify, InotifyEvent], None], exclusive=True)¶: If ISDIR isn’t included in the flags, an exclusive handler won’t be executed.