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