Statistics
| Branch: | Tag: | Revision:

root / ncclient / operations / rpc.py @ 11bab021

History | View | Annotate | Download (12.2 kB)

1
# Copyright 2009 Shikhar Bhushan
2
#
3
# Licensed under the Apache License, Version 2.0 (the "License");
4
# you may not use this file except in compliance with the License.
5
# You may obtain a copy of the License at
6
#
7
#    http://www.apache.org/licenses/LICENSE-2.0
8
#
9
# Unless required by applicable law or agreed to in writing, software
10
# distributed under the License is distributed on an "AS IS" BASIS,
11
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
# See the License for the specific language governing permissions and
13
# limitations under the License.
14

    
15
from threading import Event, Lock
16
from uuid import uuid1
17

    
18
from ncclient.xml_ import *
19
from ncclient.transport import SessionListener
20

    
21
from errors import OperationError, TimeoutExpiredError, MissingCapabilityError
22

    
23
import logging
24
logger = logging.getLogger("ncclient.operations.rpc")
25

    
26

    
27
class RPCReply:
28

    
29
    """Represents an *<rpc-reply>*. Only concerns itself with whether the
30
    operation was successful.
31

32
    .. note::
33
        If the reply has not yet been parsed there is an implicit, one-time
34
        parsing overhead to accessing the attributes defined by this class and
35
        any subclasses.
36
    """
37

    
38
    def __init__(self, raw):
39
        self._raw = raw
40
        self._parsed = False
41
        self._root = None
42
        self._errors = []
43

    
44
    def __repr__(self):
45
        return self._raw
46

    
47
    def _parsing_hook(self, root):
48
        """Subclass can implement.
49

50
        :type root: :class:`~xml.etree.ElementTree.Element`
51
        """
52
        pass
53

    
54
    def parse(self):
55
        """Parse the *<rpc-reply>*"""
56
        if self._parsed:
57
            return
58
        root = self._root = to_ele(self._raw) # The <rpc-reply> element
59
        # Per RFC 4741 an <ok/> tag is sent when there are no errors or warnings
60
        ok = root.find(qualify("ok"))
61
        if ok is None:
62
            # Create RPCError objects from <rpc-error> elements
63
            error = root.find(qualify("rpc-error"))
64
            if error is not None:
65
                for err in root.getiterator(error.tag):
66
                    # Process a particular <rpc-error>
67
                    self._errors.append(RPCError(err))
68
        self._parsing_hook(root)
69
        self._parsed = True
70

    
71
    @property
72
    def xml(self):
73
        "*<rpc-reply>* as returned"
74
        return self._raw
75

    
76
    @property
77
    def ok(self):
78
        "Boolean value indicating if there were no errors."
79
        if not self._parsed:
80
            self.parse()
81
        return not self._errors # empty list => false
82

    
83
    @property
84
    def error(self):
85
        """Short for :attr:`errors` [0]; :const:`None` if there were no errors.
86
        """
87
        if not self._parsed:
88
            self.parse()
89
        if self._errors:
90
            return self._errors[0]
91
        else:
92
            return None
93

    
94
    @property
95
    def errors(self):
96
        """`list` of :class:`RPCError` objects. Will be empty if there were no
97
        *<rpc-error>* elements in reply.
98
        """
99
        if not self._parsed:
100
            self.parse()
101
        return self._errors
102

    
103

    
104
class RPCError(OperationError):
105

    
106
    """Represents an *<rpc-error>*. It is a type of :exc:`OperationError`
107
    and can be raised like any other exception."""
108
    
109
    tag_to_attr = {
110
        qualify("error-type"): "_type",
111
        qualify("error-tag"): "_tag",
112
        qualify("error-severity"): "_severity",
113
        qualify("error-info"): "_info",
114
        qualify("error-path"): "_path",
115
        qualify("error-message"): "_message"
116
    }
117
    
118
    def __init__(self, err):
119
        for attr in tag_to_attr.values():
120
            setattr(self, attr, None)
121
        for subele in err:
122
            attr = tag_to_attr.get(subele.tag, None)
123
            if attr is not None:
124
                setattr(self, attr, subele.text)
125
        if self.message is not None:
126
            OperationError.__init__(self, self.message)
127
        else:
128
            OperationError.__init__(self, self.to_dict())
129
    
130
    def to_dict(self):
131
        return dict([ (attr[1:], gettattr(self, attr)) for attr in tag_to_attr.values() ])
132
    
133
    @property
134
    def type(self):
135
        "`string` representing text of *error-type* element"
136
        return self._type
137
    
138
    @property
139
    def tag(self):
140
        "`string` representing text of *error-tag* element"
141
        return self._tag
142
    
143
    @property
144
    def severity(self):
145
        "`string` representing text of *error-severity* element"
146
        return self._severity
147
    
148
    @property
149
    def path(self):
150
        "`string` or :const:`None`; representing text of *error-path* element"
151
        return self._path
152
    
153
    @property
154
    def message(self):
155
        "`string` or :const:`None`; representing text of *error-message* element"
156
        return self._message
157
    
158
    @property
159
    def info(self):
160
        "`string` (XML) or :const:`None`, representing *error-info* element"
161
        return self._info
162

    
163

    
164
class RPCReplyListener(SessionListener): # internal use
165
    
166
    creation_lock = Lock()
167
    
168
    # one instance per session -- maybe there is a better way??
169
    def __new__(cls, session):
170
        with creation_lock:
171
            instance = session.get_listener_instance(cls)
172
            if instance is None:
173
                instance = object.__new__(cls)
174
                instance._lock = Lock()
175
                instance._id2rpc = {}
176
                #instance._pipelined = session.can_pipeline
177
                session.add_listener(instance)
178
            return instance
179

    
180
    def register(self, id, rpc):
181
        with self._lock:
182
            self._id2rpc[id] = rpc
183

    
184
    def callback(self, root, raw):
185
        tag, attrs = root
186
        if tag != qualify("rpc-reply"):
187
            return
188
        for key in attrs: # in the <rpc-reply> attributes
189
            if key == "message-id": # if we found msgid attr
190
                id = attrs[key] # get the msgid
191
                with self._lock:
192
                    try:
193
                        rpc = self._id2rpc[id] # the corresponding rpc
194
                        logger.debug("Delivering to %r" % rpc)
195
                        rpc.deliver_reply(raw)
196
                    except KeyError:
197
                        raise OperationError("Unknown 'message-id': %s", id)
198
                    # no catching other exceptions, fail loudly if must
199
                    else:
200
                        # if no error delivering, can del the reference to the RPC
201
                        del self._id2rpc[id]
202
                        break
203
        else:
204
            raise OperationError("Could not find 'message-id' attribute in <rpc-reply>")
205
    
206
    def errback(self, err):
207
        try:
208
            for rpc in self._id2rpc.values():
209
                rpc.deliver_error(err)
210
        finally:
211
            self._id2rpc.clear()
212

    
213

    
214
class RPC(object):
215

    
216
    """Base class for all operations.
217

218
    Directly corresponds to *<rpc>* requests. Handles making the request, and
219
    taking delivery of the reply.
220
    """
221

    
222
    #: Subclasses can specify their dependencies on capabilities. List of URI's
223
    # or abbreviated names, e.g. ':writable-running'. These are verified at the
224
    # time of object creation. If the capability is not available, a
225
    # :exc:`MissingCapabilityError` is raised.
226
    DEPENDS = []
227

    
228
    #: Subclasses can specify a different reply class, but it must be a
229
    # subclass of :class:`RPCReply`.
230
    REPLY_CLS = RPCReply
231

    
232
    def __init__(self, session, async=False, timeout=None, raise_mode="none"):
233
        self._session = session
234
        try:
235
            for cap in self.DEPENDS:
236
                self._assert(cap)
237
        except AttributeError:
238
            pass
239
        self._async = async
240
        self._timeout = timeout
241
        self._raise_mode = raise_mode
242
        self._id = uuid1().urn # Keeps things simple instead of having a class attr that has to be locked
243
        self._listener = RPCReplyListener(session)
244
        self._listener.register(self._id, self)
245
        self._reply = None
246
        self._error = None
247
        self._event = Event()
248

    
249
    def _build(self, subele):
250
        # internal
251
        ele = new_ele("rpc", {"message-id": self._id}, xmlns=BASE_NS_1_0)
252
        ele.append(subele)
253
        return to_xml(ele)
254

    
255
    def _request(self, op):
256
        """Subclasses call this method to make the RPC request.
257
        
258
        In synchronous mode, waits until the reply is received and returns
259
        :class:`RPCReply`.
260
        
261
        In asynchronous mode, returns immediately, returning a reference to this
262
        object. The :attr:`event` attribute will be set when the reply has been
263
        received (see :attr:`reply`) or an error occured (see :attr:`error`).
264
        
265
        :type opspec: :obj:`dict` or :obj:`string` or :class:`~xml.etree.ElementTree.Element`
266
        :rtype: :class:`RPCReply` (sync) or :class:`RPC` (async)
267
        """
268
        logger.info('Requesting %r' % self.__class__.__name__)
269
        req = self._build(op)
270
        self._session.send(req)
271
        if self._async:
272
            logger.debug('Async request, returning %r', self)
273
            return self
274
        else:
275
            logger.debug('Sync request, will wait for timeout=%r' %
276
                         self._timeout)
277
            self._event.wait(self._timeout)
278
            if self._event.isSet():
279
                if self._error:
280
                    # Error that prevented reply delivery
281
                    raise self._error
282
                self._reply.parse()
283
                if self._reply.error is not None:
284
                    # <rpc-error>'s [ RPCError ]
285
                    if self._raise_mode == "all":
286
                        raise self._reply.error
287
                    elif (self._raise_mode == "errors" and
288
                          self._reply.error.type == "error"):
289
                        raise self._reply.error
290
                return self._reply
291
            else:
292
                raise TimeoutExpiredError
293

    
294
    def request(self, *args, **kwds):
295
        "Subclasses implement this method."
296
        return self._request(self.SPEC)
297
    
298
    def _assert(self, capability):
299
        """Subclasses can use this method to verify that a capability is available
300
        with the NETCONF server, before making a request that requires it. A
301
        :exc:`MissingCapabilityError` will be raised if the capability is not
302
        available."""
303
        if capability not in self._session.server_capabilities:
304
            raise MissingCapabilityError('Server does not support [%s]' %
305
                                         capability)
306
    
307
    def deliver_reply(self, raw):
308
        # internal use
309
        self._reply = self.REPLY_CLS(raw)
310
        self._event.set()
311

    
312
    def deliver_error(self, err):
313
        # internal use
314
        self._error = err
315
        self._event.set()
316

    
317
    @property
318
    def reply(self):
319
        ":class:`RPCReply` element if reply has been received or :const:`None`"
320
        return self._reply
321

    
322
    @property
323
    def error(self):
324
        """:exc:`Exception` type if an error occured or :const:`None`.
325
        
326
        .. note::
327
            This represents an error which prevented a reply from being
328
            received. An *<rpc-error>* does not fall in that category -- see
329
            :class:`RPCReply` for that.
330
        """
331
        return self._error
332

    
333
    @property
334
    def id(self):
335
        "The *message-id* for this RPC"
336
        return self._id
337

    
338
    @property
339
    def session(self):
340
        """The :class:`~ncclient.transport.Session` object associated with this
341
        RPC"""
342
        return self._session
343

    
344
    @property
345
    def event(self):
346
        """:class:`~threading.Event` that is set when reply has been received or
347
        error occured."""
348
        return self._event
349

    
350
    def set_async(self, async=True):
351
        """Set asynchronous mode for this RPC."""
352
        self._async = async
353
        if async and not session.can_pipeline:
354
            raise UserWarning('Asynchronous mode not supported for this device/session')
355

    
356
    def set_raise_mode(self, mode):
357
        assert(choice in ("all", "errors", "none"))
358
        self._raise_mode = mode
359

    
360
    def set_timeout(self, timeout):
361
        """Set the timeout for synchronous waiting; defining how long the RPC
362
        request will block on a reply before raising an error. Irrelevant for
363
        asynchronous usage."""
364
        self._timeout = timeout
365

    
366
    #: Whether this RPC is asynchronous
367
    is_async = property(fget=lambda self: self._async, fset=set_async)
368

    
369
    #: Timeout for synchronous waiting
370
    timeout = property(fget=lambda self: self._timeout, fset=set_timeout)