/
handler.py
139 lines (106 loc) · 4.04 KB
/
handler.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
127
128
129
130
131
132
133
134
135
136
137
138
139
import inspect
import typing
from contextvars import ContextVar
from dataclasses import dataclass
ctx_data = ContextVar('ctx_handler_data')
current_handler = ContextVar('current_handler')
@dataclass
class FilterObj:
filter: callable
kwargs: dict
is_async: bool
class SkipHandler(Exception):
pass
class CancelHandler(Exception):
pass
def _get_spec(func: callable):
while hasattr(func, '__wrapped__'): # Try to resolve decorated callbacks
func = func.__wrapped__
return inspect.getfullargspec(func)
def _check_spec(spec: inspect.FullArgSpec, kwargs: dict):
if spec.varkw:
return kwargs
return {k: v for k, v in kwargs.items() if k in set(spec.args + spec.kwonlyargs)}
class Handler:
def __init__(self, dispatcher, once=True, middleware_key=None):
self.dispatcher = dispatcher
self.once = once
self.handlers: typing.List[Handler.HandlerObj] = []
self.middleware_key = middleware_key
def register(self, handler, filters=None, index=None):
"""
Register callback
Filters can be awaitable or not.
:param handler: coroutine
:param filters: list of filters
:param index: you can reorder handlers
"""
from .filters import get_filters_spec
spec = _get_spec(handler)
if filters and not isinstance(filters, (list, tuple, set)):
filters = [filters]
filters = get_filters_spec(self.dispatcher, filters)
record = Handler.HandlerObj(handler=handler, spec=spec, filters=filters)
if index is None:
self.handlers.append(record)
else:
self.handlers.insert(index, record)
def unregister(self, handler):
"""
Remove handler
:param handler: callback
:return:
"""
for handler_obj in self.handlers:
if handler == handler_obj or handler == handler_obj.handler:
self.handlers.remove(handler_obj)
return True
raise ValueError('This handler is not registered!')
async def notify(self, *args):
"""
Notify handlers
:param args:
:return:
"""
from .filters import check_filters, FilterNotPassed
results = []
data = {}
ctx_data.set(data)
if self.middleware_key:
try:
await self.dispatcher.middleware.trigger(f"pre_process_{self.middleware_key}", args + (data,))
except CancelHandler: # Allow to cancel current event
return results
try:
for handler_obj in self.handlers:
try:
data.update(await check_filters(handler_obj.filters, args))
except FilterNotPassed:
continue
else:
ctx_token = current_handler.set(handler_obj.handler)
try:
if self.middleware_key:
await self.dispatcher.middleware.trigger(f"process_{self.middleware_key}", args + (data,))
partial_data = _check_spec(handler_obj.spec, data)
response = await handler_obj.handler(*args, **partial_data)
if response is not None:
results.append(response)
if self.once:
break
except SkipHandler:
continue
except CancelHandler:
break
finally:
current_handler.reset(ctx_token)
finally:
if self.middleware_key:
await self.dispatcher.middleware.trigger(f"post_process_{self.middleware_key}",
args + (results, data,))
return results
@dataclass
class HandlerObj:
handler: callable
spec: inspect.FullArgSpec
filters: typing.Optional[typing.Iterable[FilterObj]] = None