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 RtmBackend(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 save(self):
  43                 pass
  44
  45         def load(self):
  46                 pass
  47
  48         def add_project(self, name):
  49                 rsp = self._rtm.lists.add(
  50                         timeline=self._timeline,
  51                         name=name,
  52                 )
  53                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  54                 self._lists = []
  55
  56         def set_project_name(self, projId, name):
  57                 rsp = self._rtm.lists.setName(
  58                         timeline=self._timeline,
  59                         list_id=projId,
  60                         name=name,
  61                 )
  62                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  63                 self._lists = []
  64
  65         def set_project_visibility(self, projId, visibility):
  66                 action = self._rtm.lists.unarchive if visibility else self._rtm.lists.archive
  67                 rsp = action(
  68                         timeline=self._timeline,
  69                         list_id=projId,
  70                 )
  71                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  72                 self._lists = []
  73
  74         def get_projects(self):
  75                 if len(self._lists) == 0:
  76                         self._populate_projects()
  77
  78                 for list in self._lists:
  79                         yield list
  80
  81         def get_project(self, projId):
  82                 projs = [proj for proj in self.get_projects() if projId == proj["id"]]
  83                 assert len(projs) == 1, "%r: %r / %r" % (projId, projs, self._lists)
  84                 return projs[0]
  85
  86         def lookup_project(self, projName):
  87                 """
  88                 From a project's name, returns the project's details
  89                 """
  90                 todoList = [list for list in self.get_projects() if list["name"] == projName]
  91                 assert len(todoList) == 1, "Wrong number of lists found for %s, in %r" % (projName, todoList)
  92                 return todoList[0]
  93
  94         def get_locations(self):
  95                 rsp = self._rtm.locations.getList()
  96                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  97                 locations = [
  98                         dict((
  99                                 ("name", t.name),
 100                                 ("id", t.id),
 101                                 ("longitude", t.longitude),
 102                                 ("latitude", t.latitude),
 103                                 ("address", t.address),
 104                         ))
 105                         for t in rsp.locations
 106                 ]
 107                 return locations
 108
 109         def get_tasks_with_details(self, projId):
 110                 for realProjId, taskSeries in self._get_taskseries(projId):
 111                         for task in self._get_tasks(taskSeries):
 112                                 taskId = self._pack_ids(realProjId, taskSeries.id, task.id)
 113                                 rawTaskDetails = {
 114                                         "id": taskId,
 115                                         "projId": projId,
 116                                         "name": taskSeries.name,
 117                                         "url": taskSeries.url,
 118                                         "locationId": taskSeries.location_id,
 119                                         "dueDate": task.due,
 120                                         "isCompleted": task.completed,
 121                                         "completedDate": task.completed,
 122                                         "priority": task.priority,
 123                                         "estimate": task.estimate,
 124                                         "notes": dict((
 125                                                 (note["id"], note)
 126                                                 for note in self._get_notes(taskId, taskSeries.notes)
 127                                         )),
 128                                 }
 129                                 taskDetails = self._parse_task_details(rawTaskDetails)
 130                                 yield taskDetails
 131
 132         def get_task_details(self, taskId):
 133                 projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
 134                 for task in self.get_tasks_with_details(projId):
 135                         if task["id"] == taskId:
 136                                 return task
 137                 return {}
 138
 139         def add_task(self, projId, taskName):
 140                 rsp = self._rtm.tasks.add(
 141                         timeline=self._timeline,
 142                         list_id=projId,
 143                         name=taskName,
 144                 )
 145                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 146                 seriesId = rsp.list.taskseries.id
 147                 taskId = rsp.list.taskseries.task.id
 148                 name = rsp.list.taskseries.name
 149
 150                 return self._pack_ids(projId, seriesId, taskId)
 151
 152         def set_project(self, taskId, newProjId):
 153                 projId, seriesId, taskId = self._unpack_ids(taskId)
 154                 rsp = self._rtm.tasks.moveTo(
 155                         timeline=self._timeline,
 156                         from_list_id=projId,
 157                         to_list_id=newProjId,
 158                         taskseries_id=seriesId,
 159                         task_id=taskId,
 160                 )
 161                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 162
 163         def set_name(self, taskId, name):
 164                 projId, seriesId, taskId = self._unpack_ids(taskId)
 165                 rsp = self._rtm.tasks.setName(
 166                         timeline=self._timeline,
 167                         list_id=projId,
 168                         taskseries_id=seriesId,
 169                         task_id=taskId,
 170                         name=name,
 171                 )
 172                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 173
 174         def set_duedate(self, taskId, dueDate):
 175                 assert isinstance(dueDate, toolbox.Optional), (
 176                         "Date being set too definitively: %r" % dueDate
 177                 )
 178
 179                 projId, seriesId, taskId = self._unpack_ids(taskId)
 180                 rsp = self._rtm.tasks.setDueDate(
 181                         timeline=self._timeline,
 182                         list_id=projId,
 183                         taskseries_id=seriesId,
 184                         task_id=taskId,
 185                         due=dueDate,
 186                         parse=1,
 187                 )
 188                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 189
 190         def set_priority(self, taskId, priority):
 191                 assert isinstance(priority, toolbox.Optional), (
 192                         "Priority being set too definitively: %r" % priority
 193                 )
 194                 priority = str(priority.get_nothrow("N"))
 195                 projId, seriesId, taskId = self._unpack_ids(taskId)
 196
 197                 rsp = self._rtm.tasks.setPriority(
 198                         timeline=self._timeline,
 199                         list_id=projId,
 200                         taskseries_id=seriesId,
 201                         task_id=taskId,
 202                         priority=priority,
 203                 )
 204                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 205
 206         def complete_task(self, taskId):
 207                 projId, seriesId, taskId = self._unpack_ids(taskId)
 208
 209                 rsp = self._rtm.tasks.complete(
 210                         timeline=self._timeline,
 211                         list_id=projId,
 212                         taskseries_id=seriesId,
 213                         task_id=taskId,
 214                 )
 215                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 216
 217         def add_note(self, taskId, noteTitle, noteBody):
 218                 projId, seriesId, taskId = self._unpack_ids(taskId)
 219
 220                 rsp = self._rtm.tasks.notes.add(
 221                         timeline=self._timeline,
 222                         list_id=projId,
 223                         taskseries_id=seriesId,
 224                         task_id=taskId,
 225                         note_title=noteTitle,
 226                         note_text=noteBody,
 227                 )
 228                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 229
 230         def update_note(self, noteId, noteTitle, noteBody):
 231                 projId, seriesId, taskId, note = self._unpack_ids(noteId)
 232
 233                 rsp = self._rtm.tasks.notes.edit(
 234                         timeline=self._timeline,
 235                         note_id=noteId,
 236                         note_title=noteTitle,
 237                         note_text=noteBody,
 238                 )
 239                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 240
 241         def delete_note(self, noteId):
 242                 projId, seriesId, taskId, noteId = self._unpack_ids(noteId)
 243
 244                 rsp = self._rtm.tasks.notes.delete(
 245                         timeline=self._timeline,
 246                         note_id=noteId,
 247                 )
 248                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 249
 250         @staticmethod
 251         def _pack_ids(*ids):
 252                 """
 253                 >>> RtmBackend._pack_ids(123, 456)
 254                 '123-456'
 255                 """
 256                 return "-".join((str(id) for id in ids))
 257
 258         @staticmethod
 259         def _unpack_ids(ids):
 260                 """
 261                 >>> RtmBackend._unpack_ids("123-456")
 262                 ['123', '456']
 263                 """
 264                 return ids.split("-")
 265
 266         def _get_taskseries(self, projId):
 267                 rsp = self._rtm.tasks.getList(
 268                         list_id=projId,
 269                 )
 270                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 271                 # @note Meta-projects return lists for each project (I think)
 272                 rspTasksList = rsp.tasks.list
 273
 274                 if not isinstance(rspTasksList, list):
 275                         rspTasksList = (rspTasksList, )
 276
 277                 for something in rspTasksList:
 278                         realProjId = something.id
 279                         try:
 280                                 something.taskseries
 281                         except AttributeError:
 282                                 continue
 283
 284                         if isinstance(something.taskseries, list):
 285                                 somethingsTaskseries = something.taskseries
 286                         else:
 287                                 somethingsTaskseries = (something.taskseries, )
 288
 289                         for taskSeries in somethingsTaskseries:
 290                                 yield realProjId, taskSeries
 291
 292         def _get_tasks(self, taskSeries):
 293                 if isinstance(taskSeries.task, list):
 294                         tasks = taskSeries.task
 295                 else:
 296                         tasks = (taskSeries.task, )
 297                 for task in tasks:
 298                         yield task
 299
 300         def _parse_task_details(self, rawTaskDetails):
 301                 taskDetails = {}
 302                 taskDetails["id"] = rawTaskDetails["id"]
 303                 taskDetails["projId"] = rawTaskDetails["projId"]
 304                 taskDetails["name"] = rawTaskDetails["name"]
 305                 taskDetails["url"] = fix_url(rawTaskDetails["url"])
 306
 307                 rawLocationId = rawTaskDetails["locationId"]
 308                 if rawLocationId:
 309                         locationId = toolbox.Optional(rawLocationId)
 310                 else:
 311                         locationId = toolbox.Optional()
 312                 taskDetails["locationId"] = locationId
 313
 314                 if rawTaskDetails["dueDate"]:
 315                         dueDate = datetime.datetime.strptime(
 316                                 rawTaskDetails["dueDate"],
 317                                 "%Y-%m-%dT%H:%M:%SZ",
 318                         )
 319                         dueDate = toolbox.Optional(dueDate)
 320                 else:
 321                         dueDate = toolbox.Optional()
 322                 taskDetails["dueDate"] = dueDate
 323
 324                 taskDetails["isCompleted"] = len(rawTaskDetails["isCompleted"]) != 0
 325
 326                 if rawTaskDetails["completedDate"]:
 327                         completedDate = datetime.datetime.strptime(
 328                                 rawTaskDetails["completedDate"],
 329                                 "%Y-%m-%dT%H:%M:%SZ",
 330                         )
 331                         completedDate = toolbox.Optional(completedDate)
 332                 else:
 333                         completedDate = toolbox.Optional()
 334                 taskDetails["completedDate"] = completedDate
 335
 336                 try:
 337                         priority = toolbox.Optional(int(rawTaskDetails["priority"]))
 338                 except ValueError:
 339                         priority = toolbox.Optional()
 340                 taskDetails["priority"] = priority
 341
 342                 if rawTaskDetails["estimate"]:
 343                         estimate = rawTaskDetails["estimate"]
 344                         estimate = toolbox.Optional(estimate)
 345                 else:
 346                         estimate = toolbox.Optional()
 347                 taskDetails["estimate"] = estimate
 348
 349                 taskDetails["notes"] = rawTaskDetails["notes"]
 350
 351                 rawKeys = list(rawTaskDetails.iterkeys())
 352                 rawKeys.sort()
 353                 parsedKeys = list(taskDetails.iterkeys())
 354                 parsedKeys.sort()
 355                 assert rawKeys == parsedKeys, "Missing some, %r != %r" % (rawKeys, parsedKeys)
 356
 357                 return taskDetails
 358
 359         def _get_notes(self, taskId, notes):
 360                 if not notes:
 361                         return
 362                 elif isinstance(notes.note, list):
 363                         notes = notes.note
 364                 else:
 365                         notes = (notes.note, )
 366
 367                 projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
 368
 369                 for note in notes:
 370                         noteId = self._pack_ids(projId, rtmSeriesId, rtmTaskId, note.id)
 371                         title = note.title
 372                         body = getattr(note, "$t")
 373                         yield {
 374                                 "id": noteId,
 375                                 "title": title,
 376                                 "body": body,
 377                         }
 378
 379         def _populate_projects(self):
 380                 rsp = self._rtm.lists.getList()
 381                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 382                 del self._lists[:]
 383                 self._lists.extend((
 384                         dict((
 385                                 ("name", t.name),
 386                                 ("id", t.id),
 387                                 ("isVisible", not int(t.archived)),
 388                                 ("isMeta", not not int(t.smart)),
 389                         ))
 390                         for t in rsp.lists.list
 391                 ))