-
Notifications
You must be signed in to change notification settings - Fork 33
/
signals.py
126 lines (101 loc) · 4.44 KB
/
signals.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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
# Copyright (C) 2014-2017 New York University
# This file is part of ReproZip which is released under the Revised BSD License
# See file LICENSE for full license details.
"""Signal system.
Emitting and subscribing to these signals is the framework for the plugin
infrastructure.
"""
from __future__ import division, print_function, unicode_literals
import traceback
import warnings
from reprounzip.utils import irange, iteritems
class SignalWarning(UserWarning):
"""Warning from the Signal class.
Mainly useful for testing (to turn these to errors), however a 'signal:'
prefix is actually used in the messages because of Python bug 22543
http://bugs.python.org/issue22543
"""
class Signal(object):
"""A signal, with its set of arguments.
This holds the expected parameters that the signal expects, in several
categories:
* `expected_args` are the arguments of the signals that must be set. Trying
to emit the signal without these will show a warning and won't touch the
listeners. Listeners can rely on these being set.
* `new_args` are new arguments that listeners cannot yet rely on but that
emitters should try to pass in. Missing arguments doesn't show a warning
yet but might in the future.
* `old_args` are arguments that you might still pass in but that you should
move away from; they will show a warning stating their deprecation.
Listeners can subscribe to a signal, and may be any callable hashable
object.
"""
REQUIRED, OPTIONAL, DEPRECATED = irange(3)
def __init__(self, expected_args=[], new_args=[], old_args=[]):
self._args = {}
self._args.update((arg, Signal.REQUIRED) for arg in expected_args)
self._args.update((arg, Signal.OPTIONAL) for arg in new_args)
self._args.update((arg, Signal.DEPRECATED) for arg in old_args)
if (len(expected_args) + len(new_args) + len(old_args) !=
len(self._args)):
raise ValueError("Repeated argument names")
self._listeners = set()
def __call__(self, **kwargs):
info = {}
for arg, argtype in iteritems(self._args):
if argtype == Signal.REQUIRED:
try:
info[arg] = kwargs.pop(arg)
except KeyError:
warnings.warn("signal: Missing required argument %s; "
"signal ignored" % arg,
category=SignalWarning,
stacklevel=2)
return
else:
if arg in kwargs:
info[arg] = kwargs.pop(arg)
if argtype == Signal.DEPRECATED:
warnings.warn(
"signal: Argument %s is deprecated" % arg,
category=SignalWarning,
stacklevel=2)
if kwargs:
arg = next(iter(kwargs))
warnings.warn(
"signal: Unexpected argument %s; signal ignored" % arg,
category=SignalWarning,
stacklevel=2)
return
for listener in self._listeners:
try:
listener(**info)
except Exception:
traceback.print_exc()
warnings.warn("signal: Got an exception calling a signal",
category=SignalWarning)
def subscribe(self, func):
"""Adds the given callable to the listeners.
It must be callable and hashable (it will be put in a set).
It will be called with the signals' arguments as keywords. Because new
parameters might be introduced, it should accept these by using::
def my_listener(param1, param2, **kwargs_):
"""
if not callable(func):
raise TypeError("%r object is not callable" % type(func))
self._listeners.add(func)
def unsubscribe(self, func):
"""Removes the given callable from the listeners.
If the listener wasn't subscribed, does nothing.
"""
self._listeners.discard(func)
pre_setup = Signal(['target', 'pack'])
post_setup = Signal(['target'], ['pack'])
pre_destroy = Signal(['target'])
post_destroy = Signal(['target'])
pre_run = Signal(['target'])
post_run = Signal(['target', 'retcode'])
pre_parse_args = Signal(['parser', 'subparsers'])
post_parse_args = Signal(['args'])
application_finishing = Signal(['reason'])
unpacker = None