#
# Copyright (C) 2009 Stefan Seefeld
# All rights reserved.
# For license terms see the file COPYING.txt.
#
import xmlrpclib
import re
__all__ = ['Unauthorised', 'Item', 'Tracker']
class Unauthorised(Exception): pass
# Define a mapping from Roundup (server-side) exceptions to client exceptions.
exception_map = {}
exception_map['ValueError'] = ValueError
exception_map['TypeError'] = TypeError
exception_map['KeyError'] = KeyError
exception_map['roundup.hyperdb.DesignatorError'] = ValueError
exception_map['roundup.exceptions.UsageError'] = ValueError
exception_map['roundup.exceptions.Unauthorised'] = Unauthorised
error_pat = re.compile("<(?P[^ ]*) '(?P[^']*)'>:(?P.*$)")
class ExceptionUnmarshaller (xmlrpclib.Unmarshaller):
"""Raise server-side exceptions on the client side."""
def close(self):
# return response tuple and target method
if self._type is None or self._marks:
raise xmlrpclib.ResponseError()
if self._type == "fault":
d = self._stack[0]
m = error_pat.match(d['faultString'])
if m:
type, value = m.group('type'), m.group('value')
if type in exception_map:
raise exception_map[type](value)
raise xmlrpclib.Fault(**d)
return tuple(self._stack)
class ExceptionTransport (xmlrpclib.Transport):
def getparser(self):
unmarshaller = ExceptionUnmarshaller()
parser = xmlrpclib.ExpatParser(unmarshaller)
return parser, unmarshaller
class Server(xmlrpclib.ServerProxy):
def __init__ (self, *args, **kwargs):
# Supply our own transport
kwargs['transport'] = ExceptionTransport()
xmlrpclib.ServerProxy.__init__(self, *args, **kwargs)
class Item(object):
"""Provide convenient access to tracker items."""
def __init__(self, kwds):
self.__dict__.update(kwds)
class Tracker(object):
"""Tracker proxy.
This class wraps the XMLRPC server proxy. It allows a more convenient access
to tracker methods, using keyword arguments instead of properties embedded
into strings, and returning 'Item' instances instead of dictionaries.
Server-side exceptions are translated, so they can be caught by client-code."""
def __init__(self, url):
self.url_ = url
self.proxy = Server(url, allow_none=True)
def list(self, classname, propname=None):
return self.proxy.list(classname, propname)
def filter(self, classname, search_matches, filterspec,
sort=[], group=[]):
return Item(self.proxy.filter(classname, search_matches, filterspec,
sort, group))
def display(self, designator, **properties):
args = ['%s=%s'%item for item in properties.items()]
return Item(self.proxy.display(designator, *args))
def create(self, classname, **properties):
args = ['%s=%s'%item for item in properties.items()]
return self.proxy.create(classname, *args)
def set(self, designator, **properties):
args = ['%s=%s'%item for item in properties.items()]
return self.proxy.set(designator, *args)
def action(self, name, **properties):
args = ['%s=%s'%item for item in properties.items()]
return self.proxy.action(name, *args)