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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
|
from typing import Any, Optional, Callable
from asyncio import iscoroutinefunction, Future
from slixmpp.xmlstream import JID
APIHandler = Callable[
[Optional[JID], Optional[str], Optional[JID], Any],
Any
]
class APIWrapper(object):
"""Slixmpp API wrapper.
This class provide a shortened binding to access ``self.api`` from
plugins without having to specify the plugin name or the global
:class:`~.APIRegistry`.
"""
def __init__(self, api, name):
self.api = api
self.name = name
if name not in self.api.settings:
self.api.settings[name] = {}
def __getattr__(self, attr):
"""Curry API management commands with the API name."""
if attr == 'name':
return self.name
elif attr == 'settings':
return self.api.settings[self.name]
elif attr == 'register':
def partial(handler, op, jid=None, node=None, default=False):
register = getattr(self.api, attr)
return register(handler, self.name, op, jid, node, default)
return partial
elif attr == 'register_default':
def partial(handler, op, jid=None, node=None):
return getattr(self.api, attr)(handler, self.name, op)
return partial
elif attr in ('run', 'restore_default', 'unregister'):
def partial(*args, **kwargs):
return getattr(self.api, attr)(self.name, *args, **kwargs)
return partial
return None
def __getitem__(self, attr):
def partial(jid=None, node=None, ifrom=None, args=None):
return self.api.run(self.name, attr, jid, node, ifrom, args)
return partial
class APIRegistry(object):
"""API Registry.
This class is the global Slixmpp API registry, on which any handler will
be registed.
"""
def __init__(self, xmpp):
self._handlers = {}
self._handler_defaults = {}
self.xmpp = xmpp
self.settings = {}
def _setup(self, ctype: str, op: str):
"""Initialize the API callback dictionaries.
:param ctype: The name of the API to initialize.
:param op: The API operation to initialize.
"""
if ctype not in self.settings:
self.settings[ctype] = {}
if ctype not in self._handler_defaults:
self._handler_defaults[ctype] = {}
if ctype not in self._handlers:
self._handlers[ctype] = {}
if op not in self._handlers[ctype]:
self._handlers[ctype][op] = {'global': None,
'jid': {},
'node': {}}
def wrap(self, ctype: str) -> APIWrapper:
"""Return a wrapper object that targets a specific API."""
return APIWrapper(self, ctype)
def purge(self, ctype: str):
"""Remove all information for a given API."""
del self.settings[ctype]
del self._handler_defaults[ctype]
del self._handlers[ctype]
def run(self, ctype: str, op: str, jid: Optional[JID] = None,
node: Optional[str] = None, ifrom: Optional[JID] = None,
args: Any = None) -> Future:
"""Execute an API callback, based on specificity.
The API callback that is executed is chosen based on the combination
of the provided JID and node:
====== ======= ===================
JID node Handler
====== ======= ===================
Given Given Node + JID handler
Given None JID handler
None Given Node handler
None None Global handler
====== ======= ===================
A node handler is responsible for servicing a single node at a single
JID, while a JID handler may respond for any node at a given JID, and
the global handler will answer to any JID+node combination.
Handlers should check that the JID ``ifrom`` is authorized to perform
the desired action.
.. versionchanged:: 1.8.0
``run()`` always returns a future, if the handler is a coroutine
the future should be awaited on.
:param ctype: The name of the API to use.
:param op: The API operation to perform.
:param jid: Optionally provide specific JID.
:param node: Optionally provide specific node.
:param ifrom: Optionally provide the requesting JID.
:param args: Optional arguments to the handler.
"""
self._setup(ctype, op)
if not jid:
jid = self.xmpp.boundjid
elif jid and not isinstance(jid, JID):
jid = JID(jid)
elif jid == JID(''):
jid = self.xmpp.boundjid
if node is None:
node = ''
if self.xmpp.is_component:
if self.settings[ctype].get('component_bare', False):
jid = jid.bare
else:
jid = jid.full
else:
if self.settings[ctype].get('client_bare', False):
jid = jid.bare
else:
jid = jid.full
jid = JID(jid)
handler = self._handlers[ctype][op]['node'].get((jid, node), None)
if handler is None:
handler = self._handlers[ctype][op]['jid'].get(jid, None)
if handler is None:
handler = self._handlers[ctype][op].get('global', None)
if handler:
try:
if iscoroutinefunction(handler):
return self.xmpp.wrap(handler(jid, node, ifrom, args))
else:
future: Future = Future()
result = handler(jid, node, ifrom, args)
future.set_result(result)
return future
except TypeError:
# To preserve backward compatibility, drop the ifrom
# parameter for existing handlers that don't understand it.
return handler(jid, node, args)
def register(self, handler: APIHandler, ctype: str, op: str,
jid: Optional[JID] = None, node: Optional[str] = None,
default: bool = False):
"""Register an API callback, with JID+node specificity.
The API callback can later be executed based on the
specificity of the provided JID+node combination.
See :meth:`~.APIRegistry.run` for more details.
:param ctype: The name of the API to use.
:param op: The API operation to perform.
:param jid: Optionally provide specific JID.
:param node: Optionally provide specific node.
"""
self._setup(ctype, op)
if jid is None and node is None:
if handler is None:
handler = self._handler_defaults[op]
self._handlers[ctype][op]['global'] = handler
elif jid is not None and node is None:
self._handlers[ctype][op]['jid'][jid] = handler
else:
self._handlers[ctype][op]['node'][(jid, node)] = handler
if default:
self.register_default(handler, ctype, op)
def register_default(self, handler, ctype: str, op: str):
"""Register a default, global handler for an operation.
:param handler: The default, global handler for the operation.
:param ctype: The name of the API to modify.
:param op: The API operation to use.
"""
self._setup(ctype, op)
self._handler_defaults[ctype][op] = handler
def unregister(self, ctype: str, op: str, jid: Optional[JID] = None,
node: Optional[str] = None):
"""Remove an API callback.
The API callback chosen for removal is based on the
specificity of the provided JID+node combination.
See :meth:`~ApiRegistry.run` for more details.
:param ctype: The name of the API to use.
:param op: The API operation to perform.
:param jid: Optionally provide specific JID.
:param node: Optionally provide specific node.
"""
self._setup(ctype, op)
self.register(None, ctype, op, jid, node)
def restore_default(self, ctype: str, op: str, jid: Optional[JID] = None,
node: Optional[str] = None):
"""Reset an API callback to use a default handler.
:param ctype: The name of the API to use.
:param op: The API operation to perform.
:param jid: Optionally provide specific JID.
:param node: Optionally provide specific node.
"""
self.unregister(ctype, op, jid, node)
self.register(self._handler_defaults[ctype][op], ctype, op, jid, node)
|
