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": dict((
  96                                                 (note["id"], note)
  97                                                 for note in self._get_notes(taskId, taskSeries.notes)
  98                                         )),
  99                                 }
 100                                 taskDetails = self._parse_task_details(rawTaskDetails)
 101                                 yield taskDetails
 102
 103         def get_task_details(self, taskId):
 104                 projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
 105                 for task in self.get_tasks_with_details(projId):
 106                         if task["id"] == taskId:
 107                                 return task
 108                 return {}
 109
 110         def add_task(self, projId, taskName):
 111                 rsp = self._rtm.tasks.add(
 112                         timeline=self._timeline,
 113                         list_id=projId,
 114                         name=taskName,
 115                 )
 116                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 117                 seriesId = rsp.list.taskseries.id
 118                 taskId = rsp.list.taskseries.task.id
 119                 name = rsp.list.taskseries.name
 120
 121                 return self._pack_ids(projId, seriesId, taskId)
 122
 123         def set_project(self, taskId, newProjId):
 124                 projId, seriesId, taskId = self._unpack_ids(taskId)
 125                 rsp = self._rtm.tasks.moveTo(
 126                         timeline=self._timeline,
 127                         from_list_id=projId,
 128                         to_list_id=newProjId,
 129                         taskseries_id=seriesId,
 130                         task_id=taskId,
 131                 )
 132                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 133
 134         def set_name(self, taskId, name):
 135                 projId, seriesId, taskId = self._unpack_ids(taskId)
 136                 rsp = self._rtm.tasks.setName(
 137                         timeline=self._timeline,
 138                         list_id=projId,
 139                         taskseries_id=seriesId,
 140                         task_id=taskId,
 141                         name=name,
 142                 )
 143                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 144
 145         def set_duedate(self, taskId, dueDate):
 146                 assert isinstance(dueDate, toolbox.Optional), (
 147                         "Date being set too definitively: %r" % dueDate
 148                 )
 149
 150                 projId, seriesId, taskId = self._unpack_ids(taskId)
 151                 rsp = self._rtm.tasks.setDueDate(
 152                         timeline=self._timeline,
 153                         list_id=projId,
 154                         taskseries_id=seriesId,
 155                         task_id=taskId,
 156                         due=dueDate,
 157                         parse=1,
 158                 )
 159                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 160
 161         def set_priority(self, taskId, priority):
 162                 assert isinstance(priority, toolbox.Optional), (
 163                         "Priority being set too definitively: %r" % priority
 164                 )
 165                 priority = str(priority.get_nothrow("N"))
 166                 projId, seriesId, taskId = self._unpack_ids(taskId)
 167
 168                 rsp = self._rtm.tasks.setPriority(
 169                         timeline=self._timeline,
 170                         list_id=projId,
 171                         taskseries_id=seriesId,
 172                         task_id=taskId,
 173                         priority=priority,
 174                 )
 175                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 176
 177         def complete_task(self, taskId):
 178                 projId, seriesId, taskId = self._unpack_ids(taskId)
 179
 180                 rsp = self._rtm.tasks.complete(
 181                         timeline=self._timeline,
 182                         list_id=projId,
 183                         taskseries_id=seriesId,
 184                         task_id=taskId,
 185                 )
 186                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 187
 188         def add_note(self, taskId, noteTitle, noteBody):
 189                 projId, seriesId, taskId = self._unpack_ids(taskId)
 190
 191                 rsp = self._rtm.tasks.notes.add(
 192                         timeline=self._timeline,
 193                         list_id=projId,
 194                         taskseries_id=seriesId,
 195                         task_id=taskId,
 196                         note_title=noteTitle,
 197                         note_text=noteBody,
 198                 )
 199                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 200
 201         def update_note(self, noteId, noteTitle, noteBody):
 202                 projId, seriesId, taskId, note = self._unpack_ids(noteId)
 203
 204                 rsp = self._rtm.tasks.notes.edit(
 205                         timeline=self._timeline,
 206                         note_id=noteId,
 207                         note_title=noteTitle,
 208                         note_text=noteBody,
 209                 )
 210                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 211
 212         def delete_note(self, noteId):
 213                 projId, seriesId, taskId, noteId = self._unpack_ids(noteId)
 214
 215                 rsp = self._rtm.tasks.notes.delete(
 216                         timeline=self._timeline,
 217                         note_id=noteId,
 218                 )
 219                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 220
 221         @staticmethod
 222         def _pack_ids(*ids):
 223                 """
 224                 >>> RtMilkManager._pack_ids(123, 456)
 225                 '123-456'
 226                 """
 227                 return "-".join((str(id) for id in ids))
 228
 229         @staticmethod
 230         def _unpack_ids(ids):
 231                 """
 232                 >>> RtMilkManager._unpack_ids("123-456")
 233                 ['123', '456']
 234                 """
 235                 return ids.split("-")
 236
 237         def _get_taskseries(self, projId):
 238                 rsp = self._rtm.tasks.getList(
 239                         list_id=projId,
 240                 )
 241                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 242                 # @note Meta-projects return lists for each project (I think)
 243                 rspTasksList = rsp.tasks.list
 244
 245                 if not isinstance(rspTasksList, list):
 246                         rspTasksList = (rspTasksList, )
 247
 248                 for something in rspTasksList:
 249                         realProjId = something.id
 250                         try:
 251                                 something.taskseries
 252                         except AttributeError:
 253                                 continue
 254
 255                         if isinstance(something.taskseries, list):
 256                                 somethingsTaskseries = something.taskseries
 257                         else:
 258                                 somethingsTaskseries = (something.taskseries, )
 259
 260                         for taskSeries in somethingsTaskseries:
 261                                 yield realProjId, taskSeries
 262
 263         def _get_tasks(self, taskSeries):
 264                 if isinstance(taskSeries.task, list):
 265                         tasks = taskSeries.task
 266                 else:
 267                         tasks = (taskSeries.task, )
 268                 for task in tasks:
 269                         yield task
 270
 271         def _parse_task_details(self, rawTaskDetails):
 272                 taskDetails = {}
 273                 taskDetails["id"] = rawTaskDetails["id"]
 274                 taskDetails["projId"] = rawTaskDetails["projId"]
 275                 taskDetails["name"] = rawTaskDetails["name"]
 276                 taskDetails["url"] = fix_url(rawTaskDetails["url"])
 277
 278                 rawLocationId = rawTaskDetails["locationId"]
 279                 if rawLocationId:
 280                         locationId = toolbox.Optional(rawLocationId)
 281                 else:
 282                         locationId = toolbox.Optional()
 283                 taskDetails["locationId"] = locationId
 284
 285                 if rawTaskDetails["dueDate"]:
 286                         dueDate = datetime.datetime.strptime(
 287                                 rawTaskDetails["dueDate"],
 288                                 "%Y-%m-%dT%H:%M:%SZ",
 289                         )
 290                         dueDate = toolbox.Optional(dueDate)
 291                 else:
 292                         dueDate = toolbox.Optional()
 293                 taskDetails["dueDate"] = dueDate
 294
 295                 taskDetails["isCompleted"] = len(rawTaskDetails["isCompleted"]) != 0
 296
 297                 if rawTaskDetails["completedDate"]:
 298                         completedDate = datetime.datetime.strptime(
 299                                 rawTaskDetails["completedDate"],
 300                                 "%Y-%m-%dT%H:%M:%SZ",
 301                         )
 302                         completedDate = toolbox.Optional(completedDate)
 303                 else:
 304                         completedDate = toolbox.Optional()
 305                 taskDetails["completedDate"] = completedDate
 306
 307                 try:
 308                         priority = toolbox.Optional(int(rawTaskDetails["priority"]))
 309                 except ValueError:
 310                         priority = toolbox.Optional()
 311                 taskDetails["priority"] = priority
 312
 313                 if rawTaskDetails["estimate"]:
 314                         estimate = rawTaskDetails["estimate"]
 315                         estimate = toolbox.Optional(estimate)
 316                 else:
 317                         estimate = toolbox.Optional()
 318                 taskDetails["estimate"] = estimate
 319
 320                 taskDetails["notes"] = rawTaskDetails["notes"]
 321
 322                 rawKeys = list(rawTaskDetails.iterkeys())
 323                 rawKeys.sort()
 324                 parsedKeys = list(taskDetails.iterkeys())
 325                 parsedKeys.sort()
 326                 assert rawKeys == parsedKeys, "Missing some, %r != %r" % (rawKeys, parsedKeys)
 327
 328                 return taskDetails
 329
 330         def _get_notes(self, taskId, notes):
 331                 if not notes:
 332                         return
 333                 elif isinstance(notes.note, list):
 334                         notes = notes.note
 335                 else:
 336                         notes = (notes.note, )
 337
 338                 projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
 339
 340                 for note in notes:
 341                         noteId = self._pack_ids(projId, rtmSeriesId, rtmTaskId, note.id)
 342                         title = note.title
 343                         body = getattr(note, "$t")
 344                         yield {
 345                                 "id": noteId,
 346                                 "title": title,
 347                                 "body": body,
 348                         }
 349
 350         def _populate_projects(self):
 351                 rsp = self._rtm.lists.getList()
 352                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 353                 del self._lists[:]
 354                 self._lists.extend((
 355                         dict((
 356                                 ("name", t.name),
 357                                 ("id", t.id),
 358                                 ("isVisible", not int(t.archived)),
 359                                 ("isMeta", not not int(t.smart)),
 360                         ))
 361                         for t in rsp.lists.list
 362                 ))