--- /dev/null
+
+"""
+Python library for Remember The Milk API
+
+@note For help, see http://www.rememberthemilk.com/services/api/methods/
+"""
+
+import weakref
+import warnings
+import urllib
+from md5 import md5
+
+_use_simplejson = False
+try:
+ import simplejson
+ _use_simplejson = True
+except ImportError:
+ pass
+
+
+__author__ = 'Sridhar Ratnakumar <http://nearfar.org/>'
+
+SERVICE_URL = 'http://api.rememberthemilk.com/services/rest/'
+AUTH_SERVICE_URL = 'http://www.rememberthemilk.com/services/auth/'
+
+
+class RTMError(StandardError):
+ pass
+
+
+class RTMAPIError(RTMError):
+ pass
+
+
+class AuthStateMachine(object):
+ """If the state is in those setup for the machine, then return
+ the datum sent. Along the way, it is an automatic call if the
+ datum is a method.
+ """
+
+ class NoData(RTMError):
+ pass
+
+ def __init__(self, states):
+ self.states = states
+ self.data = {}
+
+ def dataReceived(self, state, datum):
+ if state not in self.states:
+ raise RTMError, "Invalid state <%s>" % state
+ self.data[state] = datum
+
+ def get(self, state):
+ if state in self.data:
+ return self.data[state]
+ else:
+ raise AuthStateMachine.NoData('No data for <%s>' % state)
+
+
+class RTMapi(object):
+
+ def __init__(self, userID, apiKey, secret, token=None):
+ self._userID = userID
+ self._apiKey = apiKey
+ self._secret = secret
+ self._authInfo = AuthStateMachine(['frob', 'token'])
+
+ # this enables one to do 'rtm.tasks.getList()', for example
+ for prefix, methods in API.items():
+ setattr(self, prefix,
+ RTMAPICategory(self, prefix, methods))
+
+ if token:
+ self._authInfo.dataReceived('token', token)
+
+ def _sign(self, params):
+ "Sign the parameters with MD5 hash"
+ pairs = ''.join(['%s%s' % (k, v) for (k, v) in sortedItems(params)])
+ return md5(self._secret+pairs).hexdigest()
+
+ @staticmethod
+ def open_url(url, queryArgs=None):
+ if queryArgs:
+ url += '?' + urllib.urlencode(queryArgs)
+ warnings.warn("Performing download of %s" % url, stacklevel=5)
+ return urllib.urlopen(url)
+
+ def get(self, **params):
+ "Get the XML response for the passed `params`."
+ params['api_key'] = self._apiKey
+ params['format'] = 'json'
+ params['api_sig'] = self._sign(params)
+
+ json = self.open_url(SERVICE_URL, params).read()
+
+ if _use_simplejson:
+ data = DottedDict('ROOT', simplejson.loads(json))
+ else:
+ data = DottedDict('ROOT', safer_eval(json))
+ rsp = data.rsp
+
+ if rsp.stat == 'fail':
+ raise RTMAPIError, 'API call failed - %s (%s)' % (
+ rsp.err.msg, rsp.err.code)
+ else:
+ return rsp
+
+ def getNewFrob(self):
+ rsp = self.get(method='rtm.auth.getFrob')
+ self._authInfo.dataReceived('frob', rsp.frob)
+ return rsp.frob
+
+ def getAuthURL(self):
+ try:
+ frob = self._authInfo.get('frob')
+ except AuthStateMachine.NoData:
+ frob = self.getNewFrob()
+
+ params = {
+ 'api_key': self._apiKey,
+ 'perms': 'delete',
+ 'frob': frob
+ }
+ params['api_sig'] = self._sign(params)
+ return AUTH_SERVICE_URL + '?' + urllib.urlencode(params)
+
+ def getToken(self):
+ frob = self._authInfo.get('frob')
+ rsp = self.get(method='rtm.auth.getToken', frob=frob)
+ self._authInfo.dataReceived('token', rsp.auth.token)
+ return rsp.auth.token
+
+
+class RTMAPICategory(object):
+ "See the `API` structure and `RTM.__init__`"
+
+ def __init__(self, rtm, prefix, methods):
+ self._rtm = weakref.ref(rtm)
+ self._prefix = prefix
+ self._methods = methods
+
+ def __getattr__(self, attr):
+ if attr not in self._methods:
+ raise AttributeError, 'No such attribute: %s' % attr
+
+ rargs, oargs = self._methods[attr]
+ if self._prefix == 'tasksNotes':
+ aname = 'rtm.tasks.notes.%s' % attr
+ else:
+ aname = 'rtm.%s.%s' % (self._prefix, attr)
+ return lambda **params: self.callMethod(
+ aname, rargs, oargs, **params
+ )
+
+ def callMethod(self, aname, rargs, oargs, **params):
+ # Sanity checks
+ for requiredArg in rargs:
+ if requiredArg not in params:
+ raise TypeError, 'Required parameter (%s) missing' % requiredArg
+
+ for param in params:
+ if param not in rargs + oargs:
+ warnings.warn('Invalid parameter (%s)' % param)
+
+ return self._rtm().get(method=aname,
+ auth_token=self._rtm()._authInfo.get('token'),
+ **params)
+
+
+def sortedItems(dictionary):
+ "Return a list of (key, value) sorted based on keys"
+ keys = dictionary.keys()
+ keys.sort()
+ for key in keys:
+ yield key, dictionary[key]
+
+
+class DottedDict(object):
+ "Make dictionary items accessible via the object-dot notation."
+
+ def __init__(self, name, dictionary):
+ self._name = name
+
+ if isinstance(dictionary, dict):
+ for key, value in dictionary.items():
+ if isinstance(value, dict):
+ value = DottedDict(key, value)
+ elif isinstance(value, (list, tuple)):
+ value = [DottedDict('%s_%d' % (key, i), item)
+ for i, item in enumerate(value)]
+ setattr(self, key, value)
+
+ def __repr__(self):
+ children = [c for c in dir(self) if not c.startswith('_')]
+ return '<dotted %s: %s>' % (
+ self._name,
+ ', '.join(children))
+
+ def __str__(self):
+ children = [(c, getattr(self, c)) for c in dir(self) if not c.startswith('_')]
+ return '{dotted %s: %s}' % (
+ self._name,
+ ', '.join(
+ ('%s: "%s"' % (k, str(v)))
+ for (k, v) in children)
+ )
+
+
+def safer_eval(string):
+ return eval(string, {}, {})
+
+
+API = {
+ 'auth': {
+ 'checkToken':
+ [('auth_token'), ()],
+ 'getFrob':
+ [(), ()],
+ 'getToken':
+ [('frob'), ()]
+ },
+ 'contacts': {
+ 'add':
+ [('timeline', 'contact'), ()],
+ 'delete':
+ [('timeline', 'contact_id'), ()],
+ 'getList':
+ [(), ()],
+ },
+ 'groups': {
+ 'add':
+ [('timeline', 'group'), ()],
+ 'addContact':
+ [('timeline', 'group_id', 'contact_id'), ()],
+ 'delete':
+ [('timeline', 'group_id'), ()],
+ 'getList':
+ [(), ()],
+ 'removeContact':
+ [('timeline', 'group_id', 'contact_id'), ()],
+ },
+ 'lists': {
+ 'add':
+ [('timeline', 'name'), ('filter'), ()],
+ 'archive':
+ [('timeline', 'list_id'), ()],
+ 'delete':
+ [('timeline', 'list_id'), ()],
+ 'getList':
+ [(), ()],
+ 'setDefaultList':
+ [('timeline'), ('list_id'), ()],
+ 'setName':
+ [('timeline', 'list_id', 'name'), ()],
+ 'unarchive':
+ [('timeline'), ('list_id'), ()],
+ },
+ 'locations': {
+ 'getList':
+ [(), ()],
+ },
+ 'reflection': {
+ 'getMethodInfo':
+ [('methodName',), ()],
+ 'getMethods':
+ [(), ()],
+ },
+ 'settings': {
+ 'getList':
+ [(), ()],
+ },
+ 'tasks': {
+ 'add':
+ [('timeline', 'name',), ('list_id', 'parse',)],
+ 'addTags':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id', 'tags'),
+ ()],
+ 'complete':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id',), ()],
+ 'delete':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id'), ()],
+ 'getList':
+ [(),
+ ('list_id', 'filter', 'last_sync')],
+ 'movePriority':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id', 'direction'),
+ ()],
+ 'moveTo':
+ [('timeline', 'from_list_id', 'to_list_id', 'taskseries_id', 'task_id'),
+ ()],
+ 'postpone':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id'),
+ ()],
+ 'removeTags':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id', 'tags'),
+ ()],
+ 'setDueDate':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id'),
+ ('due', 'has_due_time', 'parse')],
+ 'setEstimate':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id'),
+ ('estimate',)],
+ 'setLocation':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id'),
+ ('location_id',)],
+ 'setName':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id', 'name'),
+ ()],
+ 'setPriority':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id'),
+ ('priority',)],
+ 'setRecurrence':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id'),
+ ('repeat',)],
+ 'setTags':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id'),
+ ('tags',)],
+ 'setURL':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id'),
+ ('url',)],
+ 'uncomplete':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id'),
+ ()],
+ },
+ 'tasksNotes': {
+ 'add':
+ [('timeline', 'list_id', 'taskseries_id', 'task_id', 'note_title', 'note_text'), ()],
+ 'delete':
+ [('timeline', 'note_id'), ()],
+ 'edit':
+ [('timeline', 'note_id', 'note_title', 'note_text'), ()],
+ },
+ 'test': {
+ 'echo':
+ [(), ()],
+ 'login':
+ [(), ()],
+ },
+ 'time': {
+ 'convert':
+ [('to_timezone',), ('from_timezone', 'to_timezone', 'time')],
+ 'parse':
+ [('text',), ('timezone', 'dateformat')],
+ },
+ 'timelines': {
+ 'create':
+ [(), ()],
+ },
+ 'timezones': {
+ 'getList':
+ [(), ()],
+ },
+ 'transactions': {
+ 'undo':
+ [('timeline', 'transaction_id'), ()],
+ },
+}
"""
-import rtmapi
+import rtm_api
def fix_url(rturl):
self._password = password
self._token = token
- self._rtm = rtmapi.RTMapi(self._username, self.API_KEY, self.SECRET, token)
+ self._rtm = rtm_api.RTMapi(self._username, self.API_KEY, self.SECRET, token)
self._token = token
resp = self._rtm.timelines.create()
self._timeline = resp.timeline
import toolbox
import gtk_toolbox
import rtm_backend
-import rtmapi
+import rtm_api
def abbreviate(text, expectedLen):
def get_token(username, apiKey, secret):
token = None
- rtm = rtmapi.RTMapi(username, apiKey, secret, token)
+ rtm = rtm_api.RTMapi(username, apiKey, secret, token)
authURL = rtm.getAuthURL()
webbrowser.open(authURL)
self._manager = rtm_backend.RtMilkManager(*credentials)
self._credentials = credentials
return # Login succeeded
- except rtmapi.AuthStateMachine.NoData:
+ except rtm_api.AuthStateMachine.NoData:
# Login failed, grab new credentials
credentials = get_credentials(self._credentialsDialog)
+++ /dev/null
-
-"""
-Python library for Remember The Milk API
-
-@note For help, see http://www.rememberthemilk.com/services/api/methods/
-"""
-
-import weakref
-import warnings
-import urllib
-from md5 import md5
-
-_use_simplejson = False
-try:
- import simplejson
- _use_simplejson = True
-except ImportError:
- pass
-
-
-__author__ = 'Sridhar Ratnakumar <http://nearfar.org/>'
-
-SERVICE_URL = 'http://api.rememberthemilk.com/services/rest/'
-AUTH_SERVICE_URL = 'http://www.rememberthemilk.com/services/auth/'
-
-
-class RTMError(StandardError):
- pass
-
-
-class RTMAPIError(RTMError):
- pass
-
-
-class AuthStateMachine(object):
- """If the state is in those setup for the machine, then return
- the datum sent. Along the way, it is an automatic call if the
- datum is a method.
- """
-
- class NoData(RTMError):
- pass
-
- def __init__(self, states):
- self.states = states
- self.data = {}
-
- def dataReceived(self, state, datum):
- if state not in self.states:
- raise RTMError, "Invalid state <%s>" % state
- self.data[state] = datum
-
- def get(self, state):
- if state in self.data:
- return self.data[state]
- else:
- raise AuthStateMachine.NoData('No data for <%s>' % state)
-
-
-class RTMapi(object):
-
- def __init__(self, userID, apiKey, secret, token=None):
- self._userID = userID
- self._apiKey = apiKey
- self._secret = secret
- self._authInfo = AuthStateMachine(['frob', 'token'])
-
- # this enables one to do 'rtm.tasks.getList()', for example
- for prefix, methods in API.items():
- setattr(self, prefix,
- RTMAPICategory(self, prefix, methods))
-
- if token:
- self._authInfo.dataReceived('token', token)
-
- def _sign(self, params):
- "Sign the parameters with MD5 hash"
- pairs = ''.join(['%s%s' % (k, v) for (k, v) in sortedItems(params)])
- return md5(self._secret+pairs).hexdigest()
-
- @staticmethod
- def open_url(url, queryArgs=None):
- if queryArgs:
- url += '?' + urllib.urlencode(queryArgs)
- warnings.warn("Performing download of %s" % url, stacklevel=5)
- return urllib.urlopen(url)
-
- def get(self, **params):
- "Get the XML response for the passed `params`."
- params['api_key'] = self._apiKey
- params['format'] = 'json'
- params['api_sig'] = self._sign(params)
-
- json = self.open_url(SERVICE_URL, params).read()
-
- if _use_simplejson:
- data = DottedDict('ROOT', simplejson.loads(json))
- else:
- data = DottedDict('ROOT', safer_eval(json))
- rsp = data.rsp
-
- if rsp.stat == 'fail':
- raise RTMAPIError, 'API call failed - %s (%s)' % (
- rsp.err.msg, rsp.err.code)
- else:
- return rsp
-
- def getNewFrob(self):
- rsp = self.get(method='rtm.auth.getFrob')
- self._authInfo.dataReceived('frob', rsp.frob)
- return rsp.frob
-
- def getAuthURL(self):
- try:
- frob = self._authInfo.get('frob')
- except AuthStateMachine.NoData:
- frob = self.getNewFrob()
-
- params = {
- 'api_key': self._apiKey,
- 'perms': 'delete',
- 'frob': frob
- }
- params['api_sig'] = self._sign(params)
- return AUTH_SERVICE_URL + '?' + urllib.urlencode(params)
-
- def getToken(self):
- frob = self._authInfo.get('frob')
- rsp = self.get(method='rtm.auth.getToken', frob=frob)
- self._authInfo.dataReceived('token', rsp.auth.token)
- return rsp.auth.token
-
-
-class RTMAPICategory(object):
- "See the `API` structure and `RTM.__init__`"
-
- def __init__(self, rtm, prefix, methods):
- self._rtm = weakref.ref(rtm)
- self._prefix = prefix
- self._methods = methods
-
- def __getattr__(self, attr):
- if attr not in self._methods:
- raise AttributeError, 'No such attribute: %s' % attr
-
- rargs, oargs = self._methods[attr]
- if self._prefix == 'tasksNotes':
- aname = 'rtm.tasks.notes.%s' % attr
- else:
- aname = 'rtm.%s.%s' % (self._prefix, attr)
- return lambda **params: self.callMethod(
- aname, rargs, oargs, **params
- )
-
- def callMethod(self, aname, rargs, oargs, **params):
- # Sanity checks
- for requiredArg in rargs:
- if requiredArg not in params:
- raise TypeError, 'Required parameter (%s) missing' % requiredArg
-
- for param in params:
- if param not in rargs + oargs:
- warnings.warn('Invalid parameter (%s)' % param)
-
- return self._rtm().get(method=aname,
- auth_token=self._rtm()._authInfo.get('token'),
- **params)
-
-
-def sortedItems(dictionary):
- "Return a list of (key, value) sorted based on keys"
- keys = dictionary.keys()
- keys.sort()
- for key in keys:
- yield key, dictionary[key]
-
-
-class DottedDict(object):
- "Make dictionary items accessible via the object-dot notation."
-
- def __init__(self, name, dictionary):
- self._name = name
-
- if isinstance(dictionary, dict):
- for key, value in dictionary.items():
- if isinstance(value, dict):
- value = DottedDict(key, value)
- elif isinstance(value, (list, tuple)):
- value = [DottedDict('%s_%d' % (key, i), item)
- for i, item in enumerate(value)]
- setattr(self, key, value)
-
- def __repr__(self):
- children = [c for c in dir(self) if not c.startswith('_')]
- return '<dotted %s: %s>' % (
- self._name,
- ', '.join(children))
-
- def __str__(self):
- children = [(c, getattr(self, c)) for c in dir(self) if not c.startswith('_')]
- return '{dotted %s: %s}' % (
- self._name,
- ', '.join(
- ('%s: "%s"' % (k, str(v)))
- for (k, v) in children)
- )
-
-
-def safer_eval(string):
- return eval(string, {}, {})
-
-
-API = {
- 'auth': {
- 'checkToken':
- [('auth_token'), ()],
- 'getFrob':
- [(), ()],
- 'getToken':
- [('frob'), ()]
- },
- 'contacts': {
- 'add':
- [('timeline', 'contact'), ()],
- 'delete':
- [('timeline', 'contact_id'), ()],
- 'getList':
- [(), ()],
- },
- 'groups': {
- 'add':
- [('timeline', 'group'), ()],
- 'addContact':
- [('timeline', 'group_id', 'contact_id'), ()],
- 'delete':
- [('timeline', 'group_id'), ()],
- 'getList':
- [(), ()],
- 'removeContact':
- [('timeline', 'group_id', 'contact_id'), ()],
- },
- 'lists': {
- 'add':
- [('timeline', 'name'), ('filter'), ()],
- 'archive':
- [('timeline', 'list_id'), ()],
- 'delete':
- [('timeline', 'list_id'), ()],
- 'getList':
- [(), ()],
- 'setDefaultList':
- [('timeline'), ('list_id'), ()],
- 'setName':
- [('timeline', 'list_id', 'name'), ()],
- 'unarchive':
- [('timeline'), ('list_id'), ()],
- },
- 'locations': {
- 'getList':
- [(), ()],
- },
- 'reflection': {
- 'getMethodInfo':
- [('methodName',), ()],
- 'getMethods':
- [(), ()],
- },
- 'settings': {
- 'getList':
- [(), ()],
- },
- 'tasks': {
- 'add':
- [('timeline', 'name',), ('list_id', 'parse',)],
- 'addTags':
- [('timeline', 'list_id', 'taskseries_id', 'task_id', 'tags'),
- ()],
- 'complete':
- [('timeline', 'list_id', 'taskseries_id', 'task_id',), ()],
- 'delete':
- [('timeline', 'list_id', 'taskseries_id', 'task_id'), ()],
- 'getList':
- [(),
- ('list_id', 'filter', 'last_sync')],
- 'movePriority':
- [('timeline', 'list_id', 'taskseries_id', 'task_id', 'direction'),
- ()],
- 'moveTo':
- [('timeline', 'from_list_id', 'to_list_id', 'taskseries_id', 'task_id'),
- ()],
- 'postpone':
- [('timeline', 'list_id', 'taskseries_id', 'task_id'),
- ()],
- 'removeTags':
- [('timeline', 'list_id', 'taskseries_id', 'task_id', 'tags'),
- ()],
- 'setDueDate':
- [('timeline', 'list_id', 'taskseries_id', 'task_id'),
- ('due', 'has_due_time', 'parse')],
- 'setEstimate':
- [('timeline', 'list_id', 'taskseries_id', 'task_id'),
- ('estimate',)],
- 'setLocation':
- [('timeline', 'list_id', 'taskseries_id', 'task_id'),
- ('location_id',)],
- 'setName':
- [('timeline', 'list_id', 'taskseries_id', 'task_id', 'name'),
- ()],
- 'setPriority':
- [('timeline', 'list_id', 'taskseries_id', 'task_id'),
- ('priority',)],
- 'setRecurrence':
- [('timeline', 'list_id', 'taskseries_id', 'task_id'),
- ('repeat',)],
- 'setTags':
- [('timeline', 'list_id', 'taskseries_id', 'task_id'),
- ('tags',)],
- 'setURL':
- [('timeline', 'list_id', 'taskseries_id', 'task_id'),
- ('url',)],
- 'uncomplete':
- [('timeline', 'list_id', 'taskseries_id', 'task_id'),
- ()],
- },
- 'tasksNotes': {
- 'add':
- [('timeline', 'list_id', 'taskseries_id', 'task_id', 'note_title', 'note_text'), ()],
- 'delete':
- [('timeline', 'note_id'), ()],
- 'edit':
- [('timeline', 'note_id', 'note_title', 'note_text'), ()],
- },
- 'test': {
- 'echo':
- [(), ()],
- 'login':
- [(), ()],
- },
- 'time': {
- 'convert':
- [('to_timezone',), ('from_timezone', 'to_timezone', 'time')],
- 'parse':
- [('text',), ('timezone', 'dateformat')],
- },
- 'timelines': {
- 'create':
- [(), ()],
- },
- 'timezones': {
- 'getList':
- [(), ()],
- },
- 'transactions': {
- 'undo':
- [('timeline', 'transaction_id'), ()],
- },
-}
projects / doneit / commitdiff