Statistics
| Branch: | Tag: | Revision:

root / ncclient / operations / rpc.py @ 4bc8021f

History | View | Annotate | Download (13 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 RPCError(OperationError):
28

    
29
    """Represents an *rpc-error*. It is a type of :exc:`OperationError` and can be raised like any
30
    other exception."""
31
    
32
    tag_to_attr = {
33
        qualify("error-type"): "_type",
34
        qualify("error-tag"): "_tag",
35
        qualify("error-severity"): "_severity",
36
        qualify("error-info"): "_info",
37
        qualify("error-path"): "_path",
38
        qualify("error-message"): "_message"
39
    }
40
    
41
    def __init__(self, raw):
42
        self._raw = raw
43
        for attr in RPCError.tag_to_attr.values():
44
            setattr(self, attr, None)
45
        for subele in raw:
46
            attr = RPCError.tag_to_attr.get(subele.tag, None)
47
            if attr is not None:
48
                setattr(self, attr, subele.text if attr != "_info" else to_xml(subele) )
49
        if self.message is not None:
50
            OperationError.__init__(self, self.message)
51
        else:
52
            OperationError.__init__(self, self.to_dict())
53
    
54
    def to_dict(self):
55
        return dict([ (attr[1:], getattr(self, attr)) for attr in RPCError.tag_to_attr.values() ])
56
    
57
    "*rpc-error* element as returned."
58
    @property
59
    def xml(self):
60
        return self._raw
61
    
62
    @property
63
    def type(self):
64
        "`string` representing text of *error-type* element."
65
        return self._type
66
    
67
    @property
68
    def tag(self):
69
        "`string` representing text of *error-tag* element."
70
        return self._tag
71
    
72
    @property
73
    def severity(self):
74
        "`string` representing text of *error-severity* element."
75
        return self._severity
76
    
77
    @property
78
    def path(self):
79
        "`string` or :const:`None`; representing text of *error-path* element."
80
        return self._path
81
    
82
    @property
83
    def message(self):
84
        "`string` or :const:`None`; representing text of *error-message* element."
85
        return self._message
86
    
87
    @property
88
    def info(self):
89
        "`string` (XML) or :const:`None`; representing *error-info* element."
90
        return self._info
91

    
92

    
93
class RPCReply:
94

    
95
    """Represents an *rpc-reply*. Only concerns itself with whether the operation was successful.
96

97
    .. note::
98
        If the reply has not yet been parsed there is an implicit, one-time parsing overhead to
99
        accessing the attributes defined by this class and any subclasses.
100
    """
101
    
102
    ERROR_CLS = RPCError
103
    "Subclasses can specify a different error class, but it should be a subclass of `RPCError`."
104
    
105
    def __init__(self, raw):
106
        self._raw = raw
107
        self._parsed = False
108
        self._root = None
109
        self._errors = []
110

    
111
    def __repr__(self):
112
        return self._raw
113
    
114
    def parse(self):
115
        "Parses the *rpc-reply*."
116
        if self._parsed: return
117
        root = self._root = to_ele(self._raw) # The <rpc-reply> element
118
        # Per RFC 4741 an <ok/> tag is sent when there are no errors or warnings
119
        ok = root.find(qualify("ok"))
120
        if ok is None:
121
            # Create RPCError objects from <rpc-error> elements
122
            error = root.find(qualify("rpc-error"))
123
            if error is not None:
124
                for err in root.getiterator(error.tag):
125
                    # Process a particular <rpc-error>
126
                    self._errors.append(ERROR_CLS(err))
127
        self._parsed = True
128
    
129
    @property
130
    def xml(self):
131
        "*rpc-reply* element as returned."
132
        return self._raw
133
    
134
    @property
135
    def ok(self):
136
        "Boolean value indicating if there were no errors."
137
        return not self.errors # empty list => false
138
    
139
    @property
140
    def error(self):
141
        "Returns the first `RPCError` and :const:`None` if there were no errors."
142
        self.parse()
143
        if self._errors:
144
            return self._errors[0]
145
        else:
146
            return None
147
    
148
    @property
149
    def errors(self):
150
        """`list` of `RPCError` objects. Will be empty if there were no *rpc-error* elements in
151
        reply."""
152
        self.parse()
153
        return self._errors
154

    
155

    
156
class RPCReplyListener(SessionListener): # internal use
157
    
158
    creation_lock = Lock()
159
    
160
    # one instance per session -- maybe there is a better way??
161
    def __new__(cls, session):
162
        with RPCReplyListener.creation_lock:
163
            instance = session.get_listener_instance(cls)
164
            if instance is None:
165
                instance = object.__new__(cls)
166
                instance._lock = Lock()
167
                instance._id2rpc = {}
168
                #instance._pipelined = session.can_pipeline
169
                session.add_listener(instance)
170
            return instance
171

    
172
    def register(self, id, rpc):
173
        with self._lock:
174
            self._id2rpc[id] = rpc
175

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

    
205

    
206
class RPC(object):
207
    
208
    """Base class for all operations, directly corresponding to *rpc* requests. Handles making the
209
    request, and taking delivery of the reply."""
210
    
211
    DEPENDS = []
212
    """Subclasses can specify their dependencies on capabilities. List of URI's or abbreviated
213
    names, e.g. ':writable-running'. These are verified at the time of instantiation. If the
214
    capability is not available, a :exc:`MissingCapabilityError` is raised.
215
    """
216
    
217
    REPLY_CLS = RPCReply
218
    "Subclasses can specify a different reply class, but it should be a subclass of `RPCReply`."
219
    
220
    def __init__(self, session, async=False, timeout=None, raise_mode="none"):
221
        self._session = session
222
        try:
223
            for cap in self.DEPENDS:
224
                self._assert(cap)
225
        except AttributeError:
226
            pass
227
        self._async = async
228
        self._timeout = timeout
229
        self._raise_mode = raise_mode
230
        self._id = uuid1().urn # Keeps things simple instead of having a class attr with running ID that has to be locked
231
        self._listener = RPCReplyListener(session)
232
        self._listener.register(self._id, self)
233
        self._reply = None
234
        self._error = None
235
        self._event = Event()
236
    
237
    def _wrap(self, subele):
238
        # internal use
239
        ele = new_ele("rpc", {"message-id": self._id}, xmlns=BASE_NS_1_0)
240
        ele.append(subele)
241
        return to_xml(ele)
242

    
243
    def _request(self, op):
244
        """Implementations of :meth:`request` call this method to send the request and process the
245
        reply.
246
        
247
        In synchronous mode, blocks until the reply is received and returns `RPCReply`. Depending on
248
        the :attr:`raise_mode` a *rpc-error* element in the reply may lead to an :exc:`RPCError`
249
        exception.
250
        
251
        In asynchronous mode, returns immediately, returning *self*. The :attr:`event` attribute
252
        will be set when the reply has been received (see :attr:`reply`) or an error occured (see
253
        :attr:`error`).
254
        
255
        :param op: operation to be requested
256
        :type ops: `~xml.etree.ElementTree.Element`
257
        
258
        :rtype: `RPCReply` (sync) or `RPC` (async)
259
        """
260
        logger.info('Requesting %r' % self.__class__.__name__)
261
        req = self._wrap(op)
262
        self._session.send(req)
263
        if self._async:
264
            logger.debug('Async request, returning %r', self)
265
            return self
266
        else:
267
            logger.debug('Sync request, will wait for timeout=%r' %
268
                         self._timeout)
269
            self._event.wait(self._timeout)
270
            if self._event.isSet():
271
                if self._error:
272
                    # Error that prevented reply delivery
273
                    raise self._error
274
                self._reply.parse()
275
                if self._reply.error is not None:
276
                    # <rpc-error>'s [ RPCError ]
277
                    if self._raise_mode == "all":
278
                        raise self._reply.error
279
                    elif (self._raise_mode == "errors" and
280
                          self._reply.error.type == "error"):
281
                        raise self._reply.error
282
                return self._reply
283
            else:
284
                raise TimeoutExpiredError
285

    
286
    def request(self, *args, **kwds):
287
        """Subclasses must implement this method. Typically only the request needs to be built as an
288
        `~xml.etree.ElementTree.Element` and everything else can be handed off to
289
        :meth:`_request`."""
290
        pass
291
    
292
    def _assert(self, capability):
293
        """Subclasses can use this method to verify that a capability is available with the NETCONF
294
        server, before making a request that requires it. A :exc:`MissingCapabilityError` will be
295
        raised if the capability is not available."""
296
        if capability not in self._session.server_capabilities:
297
            raise MissingCapabilityError('Server does not support [%s]' %
298
                                         capability)
299
    
300
    def deliver_reply(self, raw):
301
        # internal use
302
        self._reply = self.REPLY_CLS(raw)
303
        self._event.set()
304

    
305
    def deliver_error(self, err):
306
        # internal use
307
        self._error = err
308
        self._event.set()
309
    
310
    @property
311
    def reply(self):
312
        "`RPCReply` element if reply has been received or :const:`None`"
313
        return self._reply
314
    
315
    @property
316
    def error(self):
317
        """:exc:`Exception` type if an error occured or :const:`None`.
318
        
319
        .. note::
320
            This represents an error which prevented a reply from being received. An *rpc-error*
321
            does not fall in that category -- see `RPCReply` for that.
322
        """
323
        return self._error
324
    
325
    @property
326
    def id(self):
327
        "The *message-id* for this RPC."
328
        return self._id
329
    
330
    @property
331
    def session(self):
332
        "The `~ncclient.transport.Session` object associated with this RPC."
333
        return self._session
334

    
335
    @property
336
    def event(self):
337
        """`~threading.Event` that is set when reply has been received or when an error preventing
338
        delivery of the reply occurs.
339
        """
340
        return self._event
341

    
342
    def set_async(self, async=True):
343
        self._async = async
344
        if async and not session.can_pipeline:
345
            raise UserWarning('Asynchronous mode not supported for this device/session')
346

    
347
    def set_raise_mode(self, mode):
348
        assert(choice in ("all", "errors", "none"))
349
        self._raise_mode = mode
350

    
351
    def set_timeout(self, timeout):
352
        self._timeout = timeout
353

    
354
    raise_mode = property(fget=lambda self: self._raise_mode, fset=set_raise_mode)
355
    """Depending on this exception raising mode, an *rpc-error* in the reply may be raised as
356
    :exc:`RPCError` exceptions. Valid values:
357
    
358
    * ``"all"`` -- any kind of *rpc-error* (error or warning)
359
    * ``"errors"`` -- when the *error-type* element says it is an error
360
    * ``"none"`` -- neither
361
    """
362
    
363
    is_async = property(fget=lambda self: self._async, fset=set_async)
364
    """Specifies whether this RPC will be / was requested asynchronously. By default RPC's are
365
    synchronous.
366
    """
367
    
368
    timeout = property(fget=lambda self: self._timeout, fset=set_timeout)
369
    """Timeout in seconds for synchronous waiting defining how long the RPC request will block on a
370
    reply before raising :exc:`TimeoutExpiredError`. By default there is no timeout, represented by
371
    :const:`None`.
372
    
373
    Irrelevant for asynchronous usage.
374
    """