sat_tmp: wokkel/rsm.py comparison

comparison wokkel/rsm.py @ 21:54f834e40341

tmp (wokkel/rsm): various improvments: - restored original tmp.wokkel.pubsub.PubSubService._toResponse_items - changed arguments order in RSMRequest.__init__ to have most common arguments first - added __str__ methods - better parsing/toElement - better handling of optional elements/attributes (count/index) - _toResponse_items handler RSM set elements without modifying original pubsub _toResponse_items - renamed parse to fromElement for coherency

author	Goffi <goffi@goffi.org>
date	Tue, 05 Jan 2016 23:20:20 +0100
parents	80f9a1a3d002
children	777b4e63fc8a

comparison

equal deleted inserted replaced

-:81f9b53ec7e4
+:54f834e40341
 """
 An expected RSM element has not been found.
 """
-class RSMRequest():
+class RSMRequest(object):
 """
 A Result Set Management request.
 @ivar max_: limit on the number of retrieved items.
 @itype max_: C{int} or C{unicode}
 @ivar before: ID of the element immediately following the page.
 @itype before: C{unicode}
 """
-def __init__(self, max_=10, index=None, after=None, before=None):
+def __init__(self, max_=10, after=None, before=None, index=None):
-max_ = int(max_)
+self.max = int(max_)
-assert max_ >= 0
-self.max = max_
 if index is not None:
 assert after is None and before is None
 index = int(index)
-assert index >= 0
 self.index = index
 if after is not None:
 assert before is None
-assert isinstance(after, unicode)
+assert isinstance(after, basestring)
 self.after = after
 if before is not None:
-assert isinstance(before, unicode)
+assert isinstance(before, basestring)
 self.before = before
+def __str__(self):
+return "RSM Request: max={0.max} after={0.after} before={0.before} index={0.index}".format(self)
 @classmethod
-def parse(cls, element):
+def fromElement(cls, element):
 """Parse the given request element.
 @param element: request containing a set element.
 @type element: L{domish.Element}
 try:
 set_elt = element.elements(NS_RSM, 'set').next()
 except StopIteration:
 raise RSMNotFoundError()
-request = RSMRequest()
+try:
-for elt in set_elt.elements():
+before_elt = set_elt.elements(NS_RSM, 'before').next()
-if elt.name in ('before', 'after'):
+except StopIteration:
-setattr(request, elt.name, ''.join(elt.children))
+before = None
-elif elt.name in ('max', 'index'):
+else:
-setattr(request, elt.name, int(''.join(elt.children)))
+before = unicode(before_elt)
-if request.max is None:
+try:
+after_elt = set_elt.elements(NS_RSM, 'after').next()
+except StopIteration:
+after = None
+else:
+after = unicode(after_elt)
+try:
+max_elt = set_elt.elements(NS_RSM, 'max').next()
+except StopIteration:
+# FIXME: even if it doesn't make a lot of sense without it
+#        <max/> element is not mandatory in XEP-0059
 raise RSMError("RSM request is missing its 'max' element")
+else:
-return request
+try:
+max_ = int(unicode(max_elt))
+except ValueError:
+raise RSMError("bad value for 'max' element")
+try:
+index_elt = set_elt.elements(NS_RSM, 'index').next()
+except StopIteration:
+index = None
+else:
+try:
+index = int(unicode(index_elt))
+except ValueError:
+raise RSMError("bad value for 'index' element")
+return RSMRequest(max_, after, before, index)
 def toElement(self):
 """
 Return the DOM representation of this RSM request.
 @rtype: L{domish.Element}
 """
 set_elt = domish.Element((NS_RSM, 'set'))
-set_elt.addElement('max').addContent(unicode(self.max))
+set_elt.addElement('max', content=unicode(self.max))
 if self.index is not None:
-set_elt.addElement('index').addContent(unicode(self.index))
+set_elt.addElement('index', content=unicode(self.index))
 if self.before is not None:
 if self.before == '':  # request the last page
 set_elt.addElement('before')
 else:
-set_elt.addElement('before').addContent(self.before)
+set_elt.addElement('before', content=self.before)
 if self.after is not None:
-set_elt.addElement('after').addContent(self.after)
+set_elt.addElement('after', content=self.after)
 return set_elt
 def render(self, element):
 """Embed the DOM representation of this RSM request in the given element.
 element.addChild(set_elt)
 return set_elt
-class RSMResponse():
+class RSMResponse(object):
 """
 A Result Set Management response.
+@ivar first: ID of the first element of the returned page.
+@itype first: C{unicode}
+@ivar last: ID of the last element of the returned page.
+@itype last: C{unicode}
+@ivar index: starting index of the returned page.
+@itype index: C{int}
 @ivar count: total number of items.
 @itype count: C{int}
-@ivar index: starting index of the returned page.
+"""
-@itype index: C{int}
+def __init__(self, first=None, last=None, index=None, count=None):
-@ivar first: ID of the first element of the returned page.
+if first is None:
-@itype first: C{unicode}
+assert last is None and index is None
+if last is None:
-@ivar last: ID of the last element of the returned page.
+assert first is None
-@itype last: C{unicode}
-"""
-def __init__(self, count=0, index=None, first=None, last=None):
-assert isinstance(count, int) and count >= 0
-self.count = count
-if index is not None:
-assert isinstance(index, int) and index >= 0
-assert isinstance(first, unicode)
-assert isinstance(last, unicode)
-else:
-assert first is None and last is None
-self.index = index
 self.first = first
 self.last = last
+if count is not None:
+self.count = int(count)
+else:
+self.count = None
+if index is not None:
+self.index = int(index)
+else:
+self.index = None
+def __str__(self):
+return "RSM Request: first={0.first} last={0.last} index={0.index} count={0.count}".format(self)
 @classmethod
-def parse(cls, element):
+def fromElement(cls, element):
 """Parse the given response element.
 @param element: response element.
 @type element: L{domish.Element}
 try:
 set_elt = element.elements(NS_RSM, 'set').next()
 except StopIteration:
 raise RSMNotFoundError()
-response = RSMResponse()
+try:
-for elt in list(set_elt.elements()):
+first_elt = set_elt.elements(NS_RSM, 'first').next()
-if elt.name in ('first', 'last'):
+except StopIteration:
-setattr(response, elt.name, ''.join(elt.children))
+first = None
-if elt.name == 'first':
+index = None
-response.index = int(elt.getAttribute("index"))
+else:
-elif elt.name == 'count':
+first = unicode(first_elt)
-response.count = int(''.join(elt.children))
+try:
+index = int(first_elt['index'])
-if response.count is None:
+except KeyError:
-raise RSMError("RSM response is missing its 'count' element")
+index = None
+except ValueError:
-return response
+raise RSMError("bad index in RSM response")
+try:
+last_elt = set_elt.elements(NS_RSM, 'last').next()
+except StopIteration:
+if first is not None:
+raise RSMError("RSM response is missing its 'last' element")
+else:
+last = None
+else:
+if first is None:
+raise RSMError("RSM response is missing its 'first' element")
+last = unicode(last_elt)
+try:
+count_elt = set_elt.elements(NS_RSM, 'count').next()
+except StopIteration:
+count = None
+else:
+try:
+count = int(unicode(count_elt))
+except ValueError:
+raise RSMError("invalid count in RSM response")
+return RSMResponse(first, last, index, count)
 def toElement(self):
 """
 Return the DOM representation of this RSM request.
 @rtype: L{domish.Element}
 """
 set_elt = domish.Element((NS_RSM, 'set'))
-set_elt.addElement('count').addContent(unicode(self.count))
+if self.first is not None:
+first_elt = set_elt.addElement('first', content=self.first)
 if self.index is not None:
-first_elt = set_elt.addElement('first')
+first_elt['index'] = unicode(self.index)
-first_elt.addContent(self.first)
-first_elt['index'] = unicode(self.index)
+set_elt.addElement('last', content=self.last)
-set_elt.addElement('last').addContent(self.last)
+if self.count is not None:
+set_elt.addElement('count', content=unicode(self.count))
 return set_elt
 def render(self, element):
 """Embed the DOM representation of this RSM response in the given element.
 @return: a dict of strings.
 @rtype: C{dict} binding C{unicode} to C{unicode}
 """
 result = {}
-for attr in ('count', 'index', 'first', 'last'):
+for attr in ('first', 'last', 'index', 'count'):
 value = getattr(self, attr)
 if value is not None:
 result[attr] = unicode(value)
 return result
 self._parameters = copy.deepcopy(pubsub.PubSubRequest._parameters)
 self._parameters['items'].append('rsm')
 def _parse_rsm(self, verbElement):
 try:
-self.rsm = RSMRequest.parse(verbElement.parent)
+self.rsm = RSMRequest.fromElement(verbElement.parent)
 except RSMNotFoundError:
 self.rsm = None
 def _render_rsm(self, verbElement):
 if self.rsm:
 for element in iq.pubsub.items.elements(pubsub.NS_PUBSUB, 'item'):
 items.append(element)
 if request.rsm:
 try:
-response = RSMResponse.parse(iq.pubsub)
+response = RSMResponse.fromElement(iq.pubsub)
 if response is not None:
 self._rsm_responses[ext_data['id']] = response
 except RSMNotFoundError:  # target pubsub server doesn't support RSM
 pass
 return items
 class PubSubService(pubsub.PubSubService):
 """PubSubService extension to handle RSM."""
 _request_class = PubSubRequest
-def _toResponse_items(self, result, resource, request):
+def _toResponse_items(self, elts, resource, request):
-response = pubsub.PubSubService._toResponse_items(self, result,
+# default method only manage <item/> elements
+# but we need to add RSM set element
+rsm_elt = None
+for idx, elt in enumerate(reversed(elts)):
+if elt.name == "set" and elt.uri == NS_RSM:
+rsm_elt = elts.pop(-1-idx)
+break
+response = pubsub.PubSubService._toResponse_items(self, elts,
 resource, request)
-for item in result:
+if rsm_elt is not None:
-if isinstance(item, tuple):
+response.addChild(rsm_elt)
-# XXX: see sat_pubsub.pgsql_storage.LeafNode.getItemsById return value
-item = item[0]
-if item.name == "set":
-response.addChild(item)
-break
 return response

Mercurial > sat_tmp

comparison wokkel/rsm.py @ 21:54f834e40341