24 Commits
Author SHA1 Message Date
spacefreak 961c64e422 remove version specific Gentoo ebuild 2022-09-08 13:46:19 +02:00
spacefreak c22bd73759 add argument singlejob to TaskScheduler 2022-09-08 13:44:42 +02:00
spacefreak 0ac52b23d4 change README.md 2022-09-04 11:28:38 +02:00
spacefreak 88ce35930c change README.md 2022-09-04 11:23:17 +02:00
spacefreak 01ccd1817d fix pyinotifyd-9999.ebuild 2022-08-12 10:02:37 +02:00
spacefreak b80d71e95e change version to 0.0.7 2022-08-12 09:59:44 +02:00
spacefreak d07316990e fix setup.py 2022-08-12 09:57:00 +02:00
spacefreak 327bd919b9 cleanup 2022-08-12 09:52:52 +02:00
spacefreak a32725504d change README.md 2022-08-12 09:50:29 +02:00
spacefreak f0b578bdb8 change PyPI distribution scripts 2022-08-12 09:43:36 +02:00
spacefreak 641d467069 change README.md and default config file 2022-08-12 09:43:18 +02:00
spacefreak f71af57288 fix installation of systemd service file 2022-08-12 09:30:52 +02:00
spacefreak 085f8f1134 change ebuild to 0.0.6 and switch to stable 2022-08-12 09:22:06 +02:00
spacefreak 50d0dbef79 fix global variable handling in config 2022-08-12 09:16:27 +02:00
spacefreak 914ec8cfb9 fix dependencies in init script 2020-11-27 11:04:34 +01:00
spacefreak 769687bca6 fix log message 2020-11-26 13:39:06 +01:00
spacefreak f9ee0c92cb provide ebuild for version 0.0.5 2020-11-24 14:13:49 +01:00
spacefreak 1a9d695374 change version to 0.0.6 2020-11-24 14:13:04 +01:00
spacefreak 9794b89b5a fix TaskState class 2020-11-24 13:25:42 +01:00
spacefreak 8445ca7cb4 fix signal handling 2020-11-24 13:20:38 +01:00
spacefreak 75241f60f4 introduce beta status 2020-11-10 01:58:08 +01:00
spacefreak 99724905be change README.md 2020-11-10 01:22:13 +01:00
spacefreak 34ac9cd596 fix typo in README.md 2020-11-10 00:57:30 +01:00
spacefreak 5b589cc999 change version to 0.0.5 2020-11-10 00:55:42 +01:00
13 changed files with 88 additions and 112 deletions
-3
View File
@@ -117,6 +117,3 @@ dmypy.json
# Temporary Vim files # Temporary Vim files
.*.swp .*.swp
# config file
/config.py
+13 -5
View File
@@ -1,5 +1,5 @@
# pyinotifyd # pyinotifyd
A daemon to monitor filesystems events with inotify on Linux and run tasks like filesystem operations (copy, move or delete), a shell command or custom async python methods. A daemon for monitoring filesystem events with inotify on Linux and run tasks like filesystem operations (copy, move or delete), a shell command or custom async python methods.
It is possible to schedule tasks with a delay, which can then be canceled again in case a canceling event occurs. A useful example for this is to run tasks only if a file has not changed within a certain amount of time. It is possible to schedule tasks with a delay, which can then be canceled again in case a canceling event occurs. A useful example for this is to run tasks only if a file has not changed within a certain amount of time.
@@ -49,7 +49,7 @@ The basic idea is to instantiate one or multiple schedulers and map specific ino
pyinotifyd has different schedulers to schedule tasks with an optional delay. The advantages of using a scheduler are consistent logging and the possibility to cancel delayed tasks. Furthermore, schedulers have the ability to differentiate between files and directories. pyinotifyd has different schedulers to schedule tasks with an optional delay. The advantages of using a scheduler are consistent logging and the possibility to cancel delayed tasks. Furthermore, schedulers have the ability to differentiate between files and directories.
### TaskScheduler ### TaskScheduler
Schedule a custom python method *job* with an optional *delay* in seconds. Skip scheduling of tasks for files and/or directories according to *files* and *dirs* arguments. If there already is a scheduled task, re-schedule it with *delay*. Use *logname* in log messages. Schedule a custom python method *job* with an optional *delay* in seconds. Skip scheduling of tasks for files and/or directories according to *files* and *dirs* arguments. If there already is a scheduled task, re-schedule it with *delay*. Use *logname* in log messages. All additional modules, functions and variables that are defined in the config file and are needed within the *job*, need to be passed as dictionary to the TaskManager through *global_vars*. If you want to limit the scheduler to run only one job at a time, set *singlejob* to True.
All arguments except for *job* are optional. All arguments except for *job* are optional.
```python ```python
# Please note that pyinotifyd uses pythons asyncio for asynchronous task execution. # Please note that pyinotifyd uses pythons asyncio for asynchronous task execution.
@@ -58,6 +58,9 @@ All arguments except for *job* are optional.
# Bad: time.sleep(10) # Bad: time.sleep(10)
# Good: await asyncio.sleep(10) # Good: await asyncio.sleep(10)
import asyncio
import logging
async def custom_job(event, task_id): async def custom_job(event, task_id):
await asyncio.sleep(10) await asyncio.sleep(10)
logging.info(f"{task_id}: execute example task: {event}") logging.info(f"{task_id}: execute example task: {event}")
@@ -67,13 +70,16 @@ task_sched = TaskScheduler(
files=True, files=True,
dirs=False, dirs=False,
delay=0, delay=0,
logname="sched") logname="sched",
global_vars=globals(),
singlejob=False)
``` ```
### ShellScheduler ### ShellScheduler
Schedule a shell command *cmd*. Replace **{maskname}**, **{pathname}** and **{src_pathname}** in *cmd* with the actual values of occuring events. This scheduler is based on TaskScheduler and has the same optional arguments. Schedule a shell command *cmd*. Replace **{maskname}**, **{pathname}** and **{src_pathname}** in *cmd* with the actual values of occuring events. This scheduler is based on TaskScheduler and has the same optional arguments.
```python ```python
# Please note that **{src_pathname}** is only present for IN_MOVED_TO events and only in the case where the IN_MOVED_FROM events are watched too. # Please note that **{src_pathname}** is only present for IN_MOVED_TO events and only
# in the case where the IN_MOVED_FROM events are watched too.
# If it is not present, the command line argument will be an empty string. # If it is not present, the command line argument will be an empty string.
shell_sched = ShellScheduler( shell_sched = ShellScheduler(
cmd="/usr/local/bin/task.sh {maskname} {pathname} {src_pathname}") cmd="/usr/local/bin/task.sh {maskname} {pathname} {src_pathname}")
@@ -129,7 +135,7 @@ event_map = {
"IN_Q_OVERFLOW": None, "IN_Q_OVERFLOW": None,
"IN_UNMOUNT": Cancel(task_sched)} "IN_UNMOUNT": Cancel(task_sched)}
# It is possible to instantiate an event map with a default scheduler set for every event, # It is possible to instantiate an event map with a default scheduler set for every event
event_map = EventMap(default_sched=task_sched) event_map = EventMap(default_sched=task_sched)
``` ```
The following events are available: The following events are available:
@@ -209,6 +215,8 @@ enableSyslog(lglevel=INFO, name="daemon")
## Schedule python method for all events on files and directories ## Schedule python method for all events on files and directories
```python ```python
import logging
async def custom_job(event, task_id): async def custom_job(event, task_id):
logging.info(f"{task_id}: execute example task: {event}") logging.info(f"{task_id}: execute example task: {event}")
@@ -1,47 +0,0 @@
# Copyright 2020 Gentoo Authors
# Distributed under the terms of the GNU General Public License v2
EAPI=7
PYTHON_COMPAT=( python3_{7,8,9} )
DISTUTILS_USE_SETUPTOOLS=rdepend
SCM=""
if [ "${PV#9999}" != "${PV}" ] ; then
SCM="git-r3"
EGIT_REPO_URI="https://github.com/spacefreak86/${PN}"
fi
inherit ${SCM} distutils-r1 systemd
DESCRIPTION="Monitore filesystems events and execute Python methods or Shell commands."
HOMEPAGE="https://github.com/spacefreak86/pymodmilter"
if [ "${PV#9999}" != "${PV}" ] ; then
SRC_URI=""
KEYWORDS=""
# Needed for tests
S="${WORKDIR}/${PN}"
EGIT_CHECKOUT_DIR="${S}"
else
SRC_URI="https://github.com/spacefreak86/${PN}/archive/${PV}.tar.gz -> ${P}.tar.gz"
KEYWORDS="amd64 x86"
fi
LICENSE="GPL-3"
SLOT="0"
IUSE="systemd"
RDEPEND="dev-python/pyinotify[${PYTHON_USEDEP}]"
python_install_all() {
distutils-r1_python_install_all
dodir /etc/${PN}
insinto /etc/${PN}
newins ${PN}/misc/config.py.default config.py
use systemd && systemd_dounit ${PN}/misc/${PN}.service
newinitd ${PN}/misc/openrc/${PN}.initd ${PN}
newconfd ${PN}/misc/openrc/${PN}.confd ${PN}
}
@@ -2,7 +2,7 @@
# Distributed under the terms of the GNU General Public License v2 # Distributed under the terms of the GNU General Public License v2
EAPI=7 EAPI=7
PYTHON_COMPAT=( python3_{7,8,9} ) PYTHON_COMPAT=( python3_{8..10} )
DISTUTILS_USE_SETUPTOOLS=rdepend DISTUTILS_USE_SETUPTOOLS=rdepend
SCM="" SCM=""
+1 -1
View File
@@ -1,6 +1,6 @@
#!/bin/bash #!/bin/bash
set -e set -e
set -x
PYTHON=$(which python) PYTHON=$(which python)
script_dir=$(dirname "$(readlink -f -- "$BASH_SOURCE")") script_dir=$(dirname "$(readlink -f -- "$BASH_SOURCE")")
+1
View File
@@ -7,6 +7,7 @@ script_dir=$(dirname "$(readlink -f -- "$BASH_SOURCE")")
pkg_dir=$(realpath "${script_dir}/../..") pkg_dir=$(realpath "${script_dir}/../..")
cd "${pkg_dir}/dist" cd "${pkg_dir}/dist"
ls -la
msg="Select version to distribute (cancel with CTRL+C):" msg="Select version to distribute (cancel with CTRL+C):"
echo "${msg}" echo "${msg}"
select version in $(find . -maxdepth 1 -type f -name "pyinotifyd-*.*.*.tar.gz" -printf "%f\n" | sed "s#\.tar\.gz##g"); do select version in $(find . -maxdepth 1 -type f -name "pyinotifyd-*.*.*.tar.gz" -printf "%f\n" | sed "s#\.tar\.gz##g"); do
+12 -13
View File
@@ -36,7 +36,7 @@ from pyinotify import ProcessEvent
from pyinotifyd._install import install, uninstall from pyinotifyd._install import install, uninstall
from pyinotifyd.scheduler import TaskScheduler, Cancel from pyinotifyd.scheduler import TaskScheduler, Cancel
__version__ = "0.0.4" __version__ = "0.0.7"
def setLoglevel(loglevel, logname=None): def setLoglevel(loglevel, logname=None):
@@ -203,12 +203,12 @@ class Pyinotifyd:
config = {} config = {}
name = Pyinotifyd.name name = Pyinotifyd.name
exec("from logging import DEBUG, INFO, WARNING, ERROR, CRITICAL", exec("from logging import DEBUG, INFO, WARNING, ERROR, CRITICAL",
{}, config) config)
exec(f"from {name} import Pyinotifyd, Watch", {}, config) exec(f"from {name} import Pyinotifyd, Watch", config)
exec(f"from {name} import setLoglevel, enableSyslog", {}, config) exec(f"from {name} import setLoglevel, enableSyslog", config)
exec(f"from {name}.scheduler import *", {}, config) exec(f"from {name}.scheduler import *", config)
with open(config_file, "r") as fh: with open(config_file, "r") as fh:
exec(fh.read(), {}, config) exec(fh.read(), config)
instance = config[f"{name}"] instance = config[f"{name}"]
assert isinstance(instance, Pyinotifyd), \ assert isinstance(instance, Pyinotifyd), \
f"{name}: expected {type(Pyinotifyd)}, " \ f"{name}: expected {type(Pyinotifyd)}, " \
@@ -436,15 +436,14 @@ def main():
ch.setFormatter(formatter) ch.setFormatter(formatter)
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
for signame in ["SIGINT", "SIGTERM"]:
loop.add_signal_handler( loop.add_signal_handler(
getattr(signal, signame), signal.SIGTERM, lambda: loop.create_task(
lambda: loop.create_task( daemon.shutdown("SIGTERM")))
daemon.shutdown(signame)))
loop.add_signal_handler( loop.add_signal_handler(
getattr(signal, "SIGHUP"), signal.SIGINT, lambda: loop.create_task(
lambda: loop.create_task( daemon.shutdown("SIGINT")))
loop.add_signal_handler(
signal.SIGHUP, lambda: loop.create_task(
daemon.reload("SIGHUP", args.config, args.debug))) daemon.reload("SIGHUP", args.config, args.debug)))
daemon.start() daemon.start()
+12 -3
View File
@@ -21,14 +21,19 @@ import shutil
import sys import sys
SYSTEMD_PATH = "/lib/systemd/system" SYSTEMD_PATHS = ["/lib/systemd/system", "/usr/lib/systemd/system"]
OPENRC = "/sbin/openrc" OPENRC = "/sbin/openrc"
def _systemd_files(pkg_dir, name): def _systemd_files(pkg_dir, name):
for path in SYSTEMD_PATHS:
if os.path.isdir(path):
break
return [ return [
(f"{pkg_dir}/misc/systemd/{name}.service", (f"{pkg_dir}/misc/systemd/{name}.service",
f"{SYSTEMD_PATH}/{name}.service", True)] f"{path}/{name}.service", True)]
def _openrc_files(pkg_dir, name): def _openrc_files(pkg_dir, name):
@@ -113,7 +118,11 @@ def _check_root():
def _check_systemd(): def _check_systemd():
systemd = os.path.isdir(SYSTEMD_PATH) for path in SYSTEMD_PATHS:
systemd = os.path.isdir(path)
if systemd:
break
if systemd: if systemd:
logging.info("systemd detected") logging.info("systemd detected")
+9 -2
View File
@@ -2,14 +2,20 @@
# TaskScheduler config # # TaskScheduler config #
########################## ##########################
#import asyncio
#import logging
#
#async def custom_job(event, task_id): #async def custom_job(event, task_id):
# await asyncio.sleep(1)
# logging.info(f"{task_id}: execute example task: {event}") # logging.info(f"{task_id}: execute example task: {event}")
# #
#task_sched = TaskScheduler( #task_sched = TaskScheduler(
# job=custom_job, # job=custom_job,
# files=True, # files=True,
# dirs=False, # dirs=False,
# delay=10) # delay=10,
# global_vars=globals(),
# singlejob=False)
########################### ###########################
@@ -20,7 +26,8 @@
# cmd="/usr/local/bin/task.sh {maskname} {pathname} {src_pathname}", # cmd="/usr/local/bin/task.sh {maskname} {pathname} {src_pathname}",
# files=True, # files=True,
# dirs=False, # dirs=False,
# delay=10) # delay=10,
# singlejob=False)
################################# #################################
-5
View File
@@ -13,11 +13,6 @@ retry="SIGTERM/${shutdown_timeout}"
extra_commands="configtest reload" extra_commands="configtest reload"
depend() {
need net
before mta
}
checkconfig() { checkconfig() {
output=$(${command} ${command_args} -t 2>&1) output=$(${command} ${command_args} -t 2>&1)
ret=$? ret=$?
+35 -19
View File
@@ -25,7 +25,6 @@ import os
import re import re
import shutil import shutil
from dataclasses import dataclass
from inspect import iscoroutinefunction from inspect import iscoroutinefunction
from shlex import quote as shell_quote from shlex import quote as shell_quote
from uuid import uuid4 from uuid import uuid4
@@ -46,14 +45,14 @@ class SchedulerLogger(logging.LoggerAdapter):
class TaskScheduler: class TaskScheduler:
@dataclass
class TaskState: class TaskState:
id: str = str(uuid4()) def __init__(self, task_id=None, task=None, cancelable=True):
task: asyncio.Task = None self.id = task_id or str(uuid4())
cancelable: bool = True self.task = task
self.cancelable = cancelable
def __init__(self, job, files=True, dirs=False, delay=0, logname="sched", def __init__(self, job, files=True, dirs=False, delay=0, logname="sched",
loop=None): loop=None, global_vars={}, singlejob=False):
assert iscoroutinefunction(job), \ assert iscoroutinefunction(job), \
f"job: expected coroutine, got {type(job)}" f"job: expected coroutine, got {type(job)}"
assert isinstance(files, bool), \ assert isinstance(files, bool), \
@@ -62,6 +61,8 @@ class TaskScheduler:
f"dirs: expected {type(bool)}, got {type(dirs)}" f"dirs: expected {type(bool)}, got {type(dirs)}"
assert isinstance(delay, int), \ assert isinstance(delay, int), \
f"delay: expected {type(int)}, got {type(delay)}" f"delay: expected {type(int)}, got {type(delay)}"
assert isinstance(global_vars, dict), \
f"global_vars: expected {type(dict)}, got {type(global_vars)}"
self._job = job self._job = job
self._files = files self._files = files
@@ -69,7 +70,8 @@ class TaskScheduler:
self._delay = delay self._delay = delay
self._log = logging.getLogger((logname or __name__)) self._log = logging.getLogger((logname or __name__))
self._loop = (loop or asyncio.get_event_loop()) self._loop = (loop or asyncio.get_event_loop())
self._globals = global_vars
self._singlejob = singlejob
self._tasks = {} self._tasks = {}
self._pause = False self._pause = False
@@ -104,6 +106,9 @@ class TaskScheduler:
else: else:
self._log.info("all remainig tasks completed") self._log.info("all remainig tasks completed")
def taskindex(self, event):
return "singlejob" if self._singlejob else event.pathname
async def _run_job(self, event, task_state, restart=False): async def _run_job(self, event, task_state, restart=False):
logger = SchedulerLogger(self._log, { logger = SchedulerLogger(self._log, {
"event": event, "event": event,
@@ -111,7 +116,7 @@ class TaskScheduler:
if self._delay > 0: if self._delay > 0:
task_state.task = self._loop.create_task( task_state.task = self._loop.create_task(
asyncio.sleep(self._delay, loop=self._loop)) asyncio.sleep(self._delay))
try: try:
if restart: if restart:
prefix = "re-" prefix = "re-"
@@ -125,7 +130,14 @@ class TaskScheduler:
return return
logger.info("start task") logger.info("start task")
if self._globals:
local_vars = {"self": self,
"event": event,
"task_id": task_state.id}
task_state.task = self._loop.create_task(
eval("self._job(event, task_id)", self._globals, local_vars))
else:
task_state.task = self._loop.create_task( task_state.task = self._loop.create_task(
self._job(event, task_state.id)) self._job(event, task_state.id))
@@ -135,9 +147,10 @@ class TaskScheduler:
except asyncio.CancelledError: except asyncio.CancelledError:
logger.warning("ongoing task cancelled") logger.warning("ongoing task cancelled")
else: else:
self._log.info("task finished") logger.info("task finished")
finally: finally:
del self._tasks[event.pathname] task_index = self.taskindex(event)
del self._tasks[task_index]
async def process_event(self, event): async def process_event(self, event):
if not ((not event.dir and self._files) or if not ((not event.dir and self._files) or
@@ -145,9 +158,13 @@ class TaskScheduler:
return return
restart = False restart = False
task_index = self.taskindex(event)
try: try:
task_state = self._tasks[event.pathname] task_state = self._tasks[task_index]
except KeyError:
task_state = TaskScheduler.TaskState()
self._tasks[task_index] = task_state
else:
logger = SchedulerLogger(self._log, { logger = SchedulerLogger(self._log, {
"event": event, "event": event,
"id": task_state.id}) "id": task_state.id})
@@ -163,16 +180,13 @@ class TaskScheduler:
logger.warning("skip event due to ongoing task") logger.warning("skip event due to ongoing task")
return return
except KeyError:
task_state = TaskScheduler.TaskState()
self._tasks[event.pathname] = task_state
if not self._pause: if not self._pause:
await self._run_job(event, task_state, restart) await self._run_job(event, task_state, restart)
async def process_cancel_event(self, event): async def process_cancel_event(self, event):
try: try:
task_state = self._tasks[event.pathname] task_index = self.taskindex(event)
task_state = self._tasks[task_index]
except KeyError: except KeyError:
return return
@@ -184,7 +198,8 @@ class TaskScheduler:
task_state.task.cancel() task_state.task.cancel()
logger.info("scheduled task cancelled") logger.info("scheduled task cancelled")
task_state.task = None task_state.task = None
del self._tasks[event.pathname] logger.info(f"{task_index}")
del self._tasks[task_index]
else: else:
logger.warning("skip event due to ongoing task") logger.warning("skip event due to ongoing task")
@@ -277,7 +292,8 @@ class FileManagerRule:
class FileManagerScheduler(TaskScheduler): class FileManagerScheduler(TaskScheduler):
def __init__(self, rules, job=None, *args, **kwargs): def __init__(self, rules, job=None, *args, **kwargs):
super().__init__(*args, **kwargs, job=self._manager_job) super().__init__(
*args, **kwargs, job=self._manager_job, singlejob=False)
if not isinstance(rules, list): if not isinstance(rules, list):
rules = [rules] rules = [rules]
+1 -1
View File
@@ -18,7 +18,7 @@ setup(name = "pyinotifyd",
# 3 - Alpha # 3 - Alpha
# 4 - Beta # 4 - Beta
# 5 - Production/Stable # 5 - Production/Stable
"Development Status :: 3 - Alpha", "Development Status :: 5 - Production/Stable",
"License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)", "License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)",
"Operating System :: OS Independent", "Operating System :: OS Independent",
"Programming Language :: Python", "Programming Language :: Python",
-9
View File
@@ -1,9 +0,0 @@
#!/usr/bin/env python
import sys
import pyinotifyd
if __name__ == '__main__':
sys.exit(
pyinotifyd.main()
)