-
Notifications
You must be signed in to change notification settings - Fork 8
/
dispatcher.py
66 lines (55 loc) · 2.08 KB
/
dispatcher.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
# -*- coding: utf-8 -*-
import os
from .plugin import Plugin
from .logging import Logger
from .context import Context
class Dispatcher(object):
"""
Dispatches tasks to loaded plugins.
"""
def __init__(self, snowblock_dir):
self._log = Logger()
self._setup_context(snowblock_dir)
self._load_plugins()
def _setup_context(self, snowblock_dir):
"""
Sets up the plugin context for the specified snowblock.
:param snowblock_dir: The directory of the snowblock
:return: The plugin context
"""
path = os.path.abspath(os.path.realpath(os.path.expanduser(snowblock_dir)))
if not os.path.exists(path):
raise DispatchError("Nonexistent snowblock directory")
self._context = Context(path)
def dispatch(self, tasks):
"""
Dispatches the specified tasks to the loaded plugins.
:param tasks: The tasks to dispatch
:return: True if all tasks have been handled by the loaded plugins successfully, False otherwise
"""
success = True
for task in tasks:
for action in task:
handled = False
if action == "defaults":
self._context.set_defaults(task[action])
handled = True
for plugin in self._plugins:
if plugin.can_handle(action):
try:
success &= plugin.handle(action, task[action])
handled = True
except Exception:
self._log.error("An error was encountered while executing action \"{}\"".format(action))
if not handled:
success = False
self._log.error("Action \"{}\" not handled".format(action))
return success
def _load_plugins(self):
"""
Loads all found plugins.
:return: None
"""
self._plugins = [plugin(self._context) for plugin in Plugin.__subclasses__()]
class DispatchError(Exception):
pass