rtm.py

   1 # Python library for Remember The Milk API
   2
   3 __author__ = 'Sridhar Ratnakumar <http://nearfar.org/>'
   4
   5
   6 import new
   7 import warnings
   8 import urllib
   9 from md5 import md5
  10
  11
  12 SERVICE_URL = 'http://api.rememberthemilk.com/services/rest/'
  13 AUTH_SERVICE_URL = 'http://www.rememberthemilk.com/services/auth/'
  14
  15
  16 DEBUG = False
  17
  18 class RTMError(Exception): pass
  19
  20 class RTMAPIError(RTMError): pass
  21
  22 class AuthStateMachine(object):
  23
  24     class NoData(RTMError): pass
  25
  26     def __init__(self, states):
  27         self.states = states
  28         self.data = {}
  29
  30     def dataReceived(self, state, datum):
  31         if state not in self.states:
  32             raise RTMError, "Invalid state <%s>" % state
  33         self.data[state] = datum
  34
  35     def get(self, state):
  36         if state in self.data:
  37             return self.data[state]
  38         else:
  39             raise AuthStateMachine.NoData, 'No data for <%s>' % state
  40
  41
  42 class RTM(object):
  43
  44     def __init__(self, apiKey, secret, token=None):
  45         self.apiKey = apiKey
  46         self.secret = secret
  47         self.auth = AuthStateMachine(['frob', 'token'])
  48
  49         # this enables one to do 'rtm.tasks.getList()', for example
  50         for prefix, methods in API.items():
  51             setattr(self, prefix,
  52                     RTMAPICategory(self, prefix, methods))
  53
  54         if token:
  55             self.auth.dataReceived('token', token)
  56
  57     def _sign(self, params):
  58         "Sign the parameters with MD5 hash"
  59         pairs = ''.join(['%s%s' % (k,v) for k,v in sortedItems(params)])
  60         return md5(self.secret+pairs).hexdigest()
  61
  62     def get(self, **params):
  63         "Get the XML response for the passed `params`."
  64         params['api_key'] = self.apiKey
  65         params['format'] = 'json'
  66         params['api_sig'] = self._sign(params)
  67
  68         json = openURL(SERVICE_URL, params).read()
  69         data = dottedJSON(json)
  70         rsp = data.rsp
  71
  72         if rsp.stat == 'fail':
  73             raise RTMAPIError, 'API call failed - %s (%s)' % (
  74                 rsp.err.msg, rsp.err.code)
  75         else:
  76             return rsp
  77
  78     def getNewFrob(self):
  79         rsp = self.get(method='rtm.auth.getFrob')
  80         self.auth.dataReceived('frob', rsp.frob)
  81         return rsp.frob
  82
  83     def getAuthURL(self):
  84         try:
  85             frob = self.auth.get('frob')
  86         except AuthStateMachine.NoData:
  87             frob = self.getNewFrob()
  88
  89         params = {
  90             'api_key': self.apiKey,
  91             'perms'  : 'delete',
  92             'frob'   : frob
  93             }
  94         params['api_sig'] = self._sign(params)
  95         return AUTH_SERVICE_URL + '?' + urllib.urlencode(params)
  96
  97     def getToken(self):
  98         frob = self.auth.get('frob')
  99         rsp = self.get(method='rtm.auth.getToken', frob=frob)
 100         self.auth.dataReceived('token', rsp.auth.token)
 101         return rsp.auth.token
 102
 103 class RTMAPICategory:
 104     "See the `RTM.API` structure and `RTM.__init__`"
 105
 106     def __init__(self, rtm, prefix, methods):
 107         self.rtm = rtm
 108         self.prefix = prefix
 109         self.methods = methods
 110
 111     def __getattr__(self, attr):
 112         if attr in self.methods:
 113             rargs, oargs = self.methods[attr]
 114             name = 'rtm.%s.%s' % (self.prefix, attr)
 115             return lambda **params: self.callMethod(
 116                 name, rargs, oargs, **params)
 117         else:
 118             raise AttributeError, 'No such attribute: ' + attr
 119
 120     def callMethod(self, name, rargs, oargs, **params):
 121         # Sanity checks
 122         for requiredArg in rargs:
 123             if requiredArg not in params:
 124                 raise TypeError, 'Required parameter (%s) missing' % requiredArg
 125
 126         for param in params:
 127             if param not in rargs + oargs:
 128                 warnings.warn('Invalid parameter (%s)' % param)
 129
 130         return self.rtm.get(method=name,
 131                             auth_token=self.rtm.auth.get('token'),
 132                             **params)
 133
 134
 135
 136 # Utility functions
 137
 138 def sortedItems(dictionary):
 139     "Return a list of (key, value) sorted based on keys"
 140     keys = dictionary.keys()
 141     keys.sort()
 142     for key in keys:
 143         yield key, dictionary[key]
 144
 145 def openURL(url, queryArgs=None):
 146     if queryArgs:
 147         url = url + '?' + urllib.urlencode(queryArgs)
 148     if DEBUG:
 149         print 'URL>', url
 150     return urllib.urlopen(url)
 151
 152 class dottedDict(object):
 153
 154     def __init__(self, name, dictionary):
 155         self._name = name
 156
 157         for key, value in dictionary.items():
 158             if type(value) is dict:
 159                 value = dottedDict(key, value)
 160             elif type(value) in (list, tuple):
 161                 value = [dottedDict('<%s:%d>' % (key, i), item)
 162                          for i, item in indexed(value)]
 163             setattr(self, key, value)
 164
 165     def __repr__(self):
 166         children = [c for c in dir(self) if not c.startswith('_')]
 167         return 'JSON<%s> : %s' % (
 168             self._name,
 169             ', '.join(children))
 170
 171
 172 def safeEval(string):
 173     return eval(string, {}, {})
 174
 175 def dottedJSON(json):
 176     return dottedDict('ROOT', safeEval(json))
 177
 178 def indexed(seq):
 179     index = 0
 180     for item in seq:
 181         yield index, item
 182         index += 1
 183
 184
 185 # API spec
 186
 187 API = {
 188     'lists': {
 189         'getList':
 190             [(), ()]
 191         },
 192     'tasks': {
 193         'addTags':
 194             # [requiredArgs, optionalArgs]
 195             [('timeline', 'list_id', 'taskseries_id', 'task_id', 'tags'),
 196              ()],
 197         'getList':
 198             [(),
 199              ('list_id', 'filter', 'last_sync')]
 200         }
 201     }
 202
 203
 204 def test(apiKey, secret, token):
 205     rtm = RTM(apiKey, secret, token)
 206     # print 'Give me access here:', rtm.getAuthURL()
 207     # raw_input('Press enter once you gave access')
 208     # print 'token is', rtm.getToken()
 209
 210     rspTasks = rtm.tasks.getList(filter="due:today status:incomplete")
 211     print [t.name for t in rspTasks.tasks.list.taskseries]
 212     print rspTasks.tasks.list.id
 213
 214     rspLists = rtm.lists.getList()
 215     print rspLists.lists.list
 216     print [(x.name, x.id) for x in rspLists.lists.list]
 217
 218