vcs.maemo.org Git - doneit/blob - src/rtm_backend.py

   1 """
   2 Wrapper for Remember The Milk API
   3 """
   4
   5 import datetime
   6
   7 import toolbox
   8 import rtm_api
   9
  10
  11 def fix_url(rturl):
  12         return "/".join(rturl.split(r"\/"))
  13
  14
  15 class RtMilkManager(object):
  16         """
  17         Interface with rememberthemilk.com
  18
  19         @todo Decide upon an interface that will end up a bit less bloated
  20         @todo Add interface for task tags
  21         @todo Add interface for postponing tasks (have way for UI to specify how many days to postpone?)
  22         @todo Add interface for task recurrence
  23         @todo Add interface for task estimate
  24         @todo Add interface for task location
  25         @todo Add interface for task url
  26         @todo Add undo support
  27         """
  28         API_KEY = '71f471f7c6ecdda6def341967686fe05'
  29         SECRET = '7d3248b085f7efbe'
  30
  31         def __init__(self, username, password, token):
  32                 self._username = username
  33                 self._password = password
  34                 self._token = token
  35
  36                 self._rtm = rtm_api.RTMapi(self._username, self.API_KEY, self.SECRET, token)
  37                 self._token = token
  38                 resp = self._rtm.timelines.create()
  39                 self._timeline = resp.timeline
  40                 self._lists = []
  41
  42         def get_projects(self):
  43                 if len(self._lists) == 0:
  44                         self._populate_projects()
  45
  46                 for list in self._lists:
  47                         yield list
  48
  49         def get_project(self, projId):
  50                 projs = [proj for proj in self.get_projects() if projId == proj["id"]]
  51                 assert len(projs) == 1, "%r: %r / %r" % (projId, projs, self._lists)
  52                 return projs[0]
  53
  54         def get_project_names(self):
  55                 return (list["name"] for list in self.get_projects)
  56
  57         def lookup_project(self, projName):
  58                 """
  59                 From a project's name, returns the project's details
  60                 """
  61                 todoList = [list for list in self.get_projects() if list["name"] == projName]
  62                 assert len(todoList) == 1, "Wrong number of lists found for %s, in %r" % (projName, todoList)
  63                 return todoList[0]
  64
  65         def get_locations(self):
  66                 rsp = self._rtm.locations.getList()
  67                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  68                 locations = [
  69                         dict((
  70                                 ("name", t.name),
  71                                 ("id", t.id),
  72                                 ("longitude", t.longitude),
  73                                 ("latitude", t.latitude),
  74                                 ("address", t.address),
  75                         ))
  76                         for t in rsp.locations
  77                 ]
  78                 return locations
  79
  80         def get_tasks_with_details(self, projId):
  81                 for realProjId, taskSeries in self._get_taskseries(projId):
  82                         for task in self._get_tasks(taskSeries):
  83                                 taskId = self._pack_ids(realProjId, taskSeries.id, task.id)
  84                                 rawTaskDetails = {
  85                                         "id": taskId,
  86                                         "projId": projId,
  87                                         "name": taskSeries.name,
  88                                         "url": taskSeries.url,
  89                                         "locationId": taskSeries.location_id,
  90                                         "dueDate": task.due,
  91                                         "isCompleted": task.completed,
  92                                         "completedDate": task.completed,
  93                                         "priority": task.priority,
  94                                         "estimate": task.estimate,
  95                                         "notes": list(self._get_notes(taskId, taskSeries.notes)),
  96                                 }
  97                                 taskDetails = self._parse_task_details(rawTaskDetails)
  98                                 yield taskDetails
  99
 100         def get_task_details(self, taskId):
 101                 projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
 102                 for task in self.get_tasks_with_details(projId):
 103                         if task["id"] == taskId:
 104                                 return task
 105                 return {}
 106
 107         def add_task(self, projId, taskName):
 108                 rsp = self._rtm.tasks.add(
 109                         timeline=self._timeline,
 110                         list_id=projId,
 111                         name=taskName,
 112                 )
 113                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 114                 seriesId = rsp.list.taskseries.id
 115                 taskId = rsp.list.taskseries.task.id
 116                 name = rsp.list.taskseries.name
 117
 118                 return self._pack_ids(projId, seriesId, taskId)
 119
 120         def set_project(self, taskId, newProjId):
 121                 projId, seriesId, taskId = self._unpack_ids(taskId)
 122                 rsp = self._rtm.tasks.moveTo(
 123                         timeline=self._timeline,
 124                         from_list_id=projId,
 125                         to_list_id=newProjId,
 126                         taskseries_id=seriesId,
 127                         task_id=taskId,
 128                 )
 129                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 130
 131         def set_name(self, taskId, name):
 132                 projId, seriesId, taskId = self._unpack_ids(taskId)
 133                 rsp = self._rtm.tasks.setName(
 134                         timeline=self._timeline,
 135                         list_id=projId,
 136                         taskseries_id=seriesId,
 137                         task_id=taskId,
 138                         name=name,
 139                 )
 140                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 141
 142         def set_duedate(self, taskId, dueDate):
 143                 assert isinstance(dueDate, toolbox.Optional), (
 144                         "Date being set too definitively: %r" % dueDate
 145                 )
 146
 147                 projId, seriesId, taskId = self._unpack_ids(taskId)
 148                 rsp = self._rtm.tasks.setDueDate(
 149                         timeline=self._timeline,
 150                         list_id=projId,
 151                         taskseries_id=seriesId,
 152                         task_id=taskId,
 153                         due=dueDate,
 154                         parse=1,
 155                 )
 156                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 157
 158         def set_priority(self, taskId, priority):
 159                 assert isinstance(priority, toolbox.Optional), (
 160                         "Priority being set too definitively: %r" % priority
 161                 )
 162                 priority = str(priority.get_nothrow("N"))
 163                 projId, seriesId, taskId = self._unpack_ids(taskId)
 164
 165                 rsp = self._rtm.tasks.setPriority(
 166                         timeline=self._timeline,
 167                         list_id=projId,
 168                         taskseries_id=seriesId,
 169                         task_id=taskId,
 170                         priority=priority,
 171                 )
 172                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 173
 174         def complete_task(self, taskId):
 175                 projId, seriesId, taskId = self._unpack_ids(taskId)
 176
 177                 rsp = self._rtm.tasks.complete(
 178                         timeline=self._timeline,
 179                         list_id=projId,
 180                         taskseries_id=seriesId,
 181                         task_id=taskId,
 182                 )
 183                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 184
 185         def add_note(self, taskId, noteTitle, noteBody):
 186                 projId, seriesId, taskId = self._unpack_ids(taskId)
 187
 188                 rsp = self._rtm.tasks.notes.add(
 189                         timeline=self._timeline,
 190                         list_id=projId,
 191                         taskseries_id=seriesId,
 192                         task_id=taskId,
 193                         note_title=noteTitle,
 194                         note_text=noteBody,
 195                 )
 196                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 197
 198         def update_note(self, noteId, noteTitle, noteBody):
 199                 projId, seriesId, taskId, note = self._unpack_ids(noteId)
 200
 201                 rsp = self._rtm.tasks.notes.edit(
 202                         timeline=self._timeline,
 203                         note_id=noteId,
 204                         note_title=noteTitle,
 205                         note_text=noteBody,
 206                 )
 207                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 208
 209         def delete_note(self, noteId):
 210                 projId, seriesId, taskId, noteId = self._unpack_ids(noteId)
 211
 212                 rsp = self._rtm.tasks.notes.delete(
 213                         timeline=self._timeline,
 214                         note_id=noteId,
 215                 )
 216                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 217
 218         @staticmethod
 219         def _pack_ids(*ids):
 220                 """
 221                 >>> RtMilkManager._pack_ids(123, 456)
 222                 '123-456'
 223                 """
 224                 return "-".join((str(id) for id in ids))
 225
 226         @staticmethod
 227         def _unpack_ids(ids):
 228                 """
 229                 >>> RtMilkManager._unpack_ids("123-456")
 230                 ['123', '456']
 231                 """
 232                 return ids.split("-")
 233
 234         def _get_taskseries(self, projId):
 235                 rsp = self._rtm.tasks.getList(
 236                         list_id=projId,
 237                 )
 238                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 239                 # @note Meta-projects return lists for each project (I think)
 240                 rspTasksList = rsp.tasks.list
 241
 242                 if not isinstance(rspTasksList, list):
 243                         rspTasksList = (rspTasksList, )
 244
 245                 for something in rspTasksList:
 246                         realProjId = something.id
 247                         try:
 248                                 something.taskseries
 249                         except AttributeError:
 250                                 continue
 251
 252                         if isinstance(something.taskseries, list):
 253                                 somethingsTaskseries = something.taskseries
 254                         else:
 255                                 somethingsTaskseries = (something.taskseries, )
 256
 257                         for taskSeries in somethingsTaskseries:
 258                                 yield realProjId, taskSeries
 259
 260         def _get_tasks(self, taskSeries):
 261                 if isinstance(taskSeries.task, list):
 262                         tasks = taskSeries.task
 263                 else:
 264                         tasks = (taskSeries.task, )
 265                 for task in tasks:
 266                         yield task
 267
 268         def _parse_task_details(self, rawTaskDetails):
 269                 taskDetails = {}
 270                 taskDetails["id"] = rawTaskDetails["id"]
 271                 taskDetails["projId"] = rawTaskDetails["projId"]
 272                 taskDetails["name"] = rawTaskDetails["name"]
 273                 taskDetails["url"] = fix_url(rawTaskDetails["url"])
 274
 275                 rawLocationId = rawTaskDetails["locationId"]
 276                 if rawLocationId:
 277                         locationId = toolbox.Optional(rawLocationId)
 278                 else:
 279                         locationId = toolbox.Optional()
 280                 taskDetails["locationId"] = locationId
 281
 282                 if rawTaskDetails["dueDate"]:
 283                         dueDate = datetime.datetime.strptime(
 284                                 rawTaskDetails["dueDate"],
 285                                 "%Y-%m-%dT%H:%M:%SZ",
 286                         )
 287                         dueDate = toolbox.Optional(dueDate)
 288                 else:
 289                         dueDate = toolbox.Optional()
 290                 taskDetails["dueDate"] = dueDate
 291
 292                 taskDetails["isCompleted"] = len(rawTaskDetails["isCompleted"]) != 0
 293
 294                 if rawTaskDetails["completedDate"]:
 295                         completedDate = datetime.datetime.strptime(
 296                                 rawTaskDetails["completedDate"],
 297                                 "%Y-%m-%dT%H:%M:%SZ",
 298                         )
 299                         completedDate = toolbox.Optional(completedDate)
 300                 else:
 301                         completedDate = toolbox.Optional()
 302                 taskDetails["completedDate"] = completedDate
 303
 304                 try:
 305                         priority = toolbox.Optional(int(rawTaskDetails["priority"]))
 306                 except ValueError:
 307                         priority = toolbox.Optional()
 308                 taskDetails["priority"] = priority
 309
 310                 if rawTaskDetails["estimate"]:
 311                         estimate = rawTaskDetails["estimate"]
 312                         estimate = toolbox.Optional(estimate)
 313                 else:
 314                         estimate = toolbox.Optional()
 315                 taskDetails["estimate"] = estimate
 316
 317                 taskDetails["notes"] = rawTaskDetails["notes"]
 318
 319                 rawKeys = list(rawTaskDetails.iterkeys())
 320                 rawKeys.sort()
 321                 parsedKeys = list(taskDetails.iterkeys())
 322                 parsedKeys.sort()
 323                 assert rawKeys == parsedKeys, "Missing some, %r != %r" % (rawKeys, parsedKeys)
 324
 325                 return taskDetails
 326
 327         def _get_notes(self, taskId, notes):
 328                 if not notes:
 329                         return
 330                 elif isinstance(notes.note, list):
 331                         notes = notes.note
 332                 else:
 333                         notes = (notes.note, )
 334
 335                 projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
 336
 337                 for note in notes:
 338                         noteId = self._pack_ids(projId, rtmSeriesId, rtmTaskId, note.id)
 339                         title = note.title
 340                         body = getattr(note, "$t")
 341                         yield {
 342                                 "id": noteId,
 343                                 "title": title,
 344                                 "body": body,
 345                         }
 346
 347         def _populate_projects(self):
 348                 rsp = self._rtm.lists.getList()
 349                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 350                 del self._lists[:]
 351                 self._lists.extend((
 352                         dict((
 353                                 ("name", t.name),
 354                                 ("id", t.id),
 355                                 ("isVisible", not int(t.archived)),
 356                                 ("isMeta", not not int(t.smart)),
 357                         ))
 358                         for t in rsp.lists.list
 359                 ))