tree /f
PS G:\workspace\py\sandbox\watchdog_sample> tree /f
Folder PATH listing for volume Users
Volume serial number is 04A1-7D2F
G:.
│ app.py
│ apps.log
│ configbase.py
│ logging_debug.conf
│ watchdog_sample.py
│
├─configs
│ │ config.py
│ │ config_debug.py
│ │ default.py
│ │
│ └─__pycache__
│
├─plugins
│ │ a.py
│ │ b.py
│ │ c.py
│ │
│ └─__pycache__
│
└─__pycache__
watchdog
"""{fname}
# sample code for module : watchdog
Usage:
"""
__author__ = 'Memoru'
__version__ = '0.0.1'
__date__ = '2018-06-23'
import sys
import os
__doc__ = __doc__.format(fname=os.path.basename(__file__))
import logging.config
import configbase as base
CONF = base.CONF
ENV = base.ENV
import sys
import time
from importlib import import_module, reload
from pathlib import Path
from watchdog.events import FileSystemEvent, PatternMatchingEventHandler
from watchdog.observers import Observer
class PluginManager:
class Handler(PatternMatchingEventHandler):
def __init__(self, manager: 'PluginManager', *args, **kwargs):
super().__init__(*args, **kwargs)
self.manager = manager
def on_created(self, event: FileSystemEvent):
print(event)
if event.src_path.endswith('.py'):
self.manager.load_plugin(Path(event.src_path))
def on_modified(self, event):
print(event)
def __init__(self, path: str):
self.plugins = {}
self.path = path
self.observer = Observer()
sys.path.append(self.path)
def start(self):
self.scan_plugin()
self.observer.schedule(self.Handler(self, patterns='*.py'), self.path)
self.observer.start()
def stop(self):
self.observer.stop()
self.observer.join()
def scan_plugin(self):
for file_path in Path(self.path).glob('*.py'):
self.load_plugin(file_path)
def load_plugin(self, file_path):
module_name = file_path.stem
if module_name not in self.plugins:
self.plugins[module_name] = import_module(module_name)
print('{} loaded.'.format(module_name))
else:
self.plugins[module_name] = reload(self.plugins[module_name])
print('{} reloaded.'.format(module_name))
def main():
plugin_manager = PluginManager('plugins/')
plugin_manager.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
plugin_manager.stop()
def process():
try:
logger = logging.getLogger(__name__)
logger.debug('◇◇--Begin sample -----------------◇◇')
logger.debug(__doc__)
main()
logger.debug('◆◆--End sample -----------------◆◆')
except Exception as ex:
logger.exception(ex)
raise ex
if __name__ == '__main__':
logging.config.fileConfig('logging_debug.conf')
process()
execution
(sandbox) PS G:\workspace\py\sandbox\watchdog_sample> python .\watchdog_sample.py
[2018/06/27 07:10:43][DEBUG](watchdog_sample.py:91) ◇◇--Begin sample -----------------◇◇
[2018/06/27 07:10:43][DEBUG](watchdog_sample.py:92) watchdog_sample.py
# sample code for module : watchdog
Usage:
a loaded.
b loaded.
c loaded.