The Watchdog Extension includes the WatchdogManager, enabling applications to easily monitor, and react to, changes in filesystem paths based on the filesystem events monitoring library Watchdog.
On application startup, the Watchdog Observer is automatically started and then upon application close, the observer thread is properly stopped and joined with the parent process before exit.
Applications using Cement <3.0.8 should continue to include watchdog in their dependencies.
Platform Support
Unix/Linux
macOS
Windows
Configuration
Application Configuration Settings
This extension does not support any application level configuration settings.
Application Meta Options
This extension honors the following application meta options:
Option
Description
watchdog_paths
Hooks
This extension defines the following hooks:
watchdog_pre_start
Run first when App.watchdog.start() is called. The application object is passed as an argument. Nothing is expected in return.
watchdog_post_start
Run last when App.watchdog.start() is called. The application object is passed as an argument. Nothing is expected in return.
watchdog_pre_stop
Run first when App.watchdog.stop() is called. The application object is passed as an argument. Nothing is expected in return.
watchdog_post_stop
Run last when App.watchdog.stop() is called. The application object is passed as an argument. Nothing is expected in return.
watchdog_pre_join
Run first when App.watchdog.join() is called. The application object is passed as an argument. Nothing is expected in return.
watchdog_post_join
Run last when App.watchdog.join() is called. The application object is passed as an argument. Nothing is expected in return.
Usage
The following example uses the default WatchdogEventHandler that by default only logs all events to debug:
from time import sleep
from cement import App, CaughtSignal
from cement.ext.ext_watchdog import WatchdogEventHandler
class MyApp(App):
class Meta:
label = 'myapp'
extensions = ['watchdog']
watchdog_paths = [
('./tmp/', WatchdogEventHandler),
]
with MyApp() as app:
app.run()
try:
while True:
sleep(1)
except CaughtSignal as e:
print(e)
In the above example, nothing is printed to console. However you will see something like the following via debug logging:
To expand on the above example, we can add our own event handlers:
myapp.py
from time import sleep
from cement import App, CaughtSignal
from cement.ext.ext_watchdog import WatchdogEventHandler
class MyEventHandler(WatchdogEventHandler):
def on_any_event(self, event):
# do something with the ``event`` object
print("The modified path was: %s" % event.src_path)
class MyApp(App):
class Meta:
label = 'myapp'
extensions = ['watchdog']
watchdog_paths = [
('./tmp/', MyEventHandler),
]
with MyApp() as app:
app.run()
try:
while True:
sleep(1)
except CaughtSignal as e:
print(e)
$ python myapp.py
The modified path was: /path/to/tmp/test.file
Note that the WatchdogEventHandler could be replaced with any other event handler classe (i.e. those available from watchdog directly). However to play nicely with Cement, we sub-class them first in order to pass in our application object.
from watchdog.events import FileSystemEventHandler
class MyEventHandler(FileSystemEventHandler):
def __init__(self, app, *args, **kw):
super(MyEventHandler, self).__init__(*args, **kw)
self.app = app