Wrapper for Remember The Milk API
"""
+import datetime
-import rtmapi
+import toolbox
+import rtm_api
def fix_url(rturl):
@todo Add interface for task estimate
@todo Add interface for task location
@todo Add interface for task url
- @todo Add interface for task notes
@todo Add undo support
"""
API_KEY = '71f471f7c6ecdda6def341967686fe05'
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
yield list
def get_project(self, projId):
- proj = [proj for proj in self.get_projects() if projId == proj["id"]]
- assert len(proj) == 1, "%r / %r" % (proj, self._lists)
- return proj[0]
+ projs = [proj for proj in self.get_projects() if projId == proj["id"]]
+ assert len(projs) == 1, "%r: %r / %r" % (projId, projs, self._lists)
+ return projs[0]
def get_project_names(self):
return (list["name"] for list in self.get_projects)
for realProjId, taskSeries in self._get_taskseries(projId):
for task in self._get_tasks(taskSeries):
taskId = self._pack_ids(realProjId, taskSeries.id, task.id)
- priority = task.priority if task.priority != "N" else ""
- yield {
+ rawTaskDetails = {
"id": taskId,
"projId": projId,
"name": taskSeries.name,
- "url": fix_url(taskSeries.url),
+ "url": taskSeries.url,
"locationId": taskSeries.location_id,
"dueDate": task.due,
- "isCompleted": len(task.completed) != 0,
- "completedDate": task.completed,
- "priority": priority,
- "estimate": task.estimate,
- "notes": list(self._get_notes(taskSeries.notes)),
- }
-
- def get_task_details(self, taskId):
- projId, seriesId, taskId = self._unpack_ids(taskId)
- for realProjId, taskSeries in self._get_taskseries(projId):
- assert projId == realProjId, "%s != %s, looks like we let leak a metalist id when packing a task id" % (projId, realProjId)
- curSeriesId = taskSeries.id
- if seriesId != curSeriesId:
- continue
- for task in self._get_tasks(taskSeries):
- curTaskId = task.id
- if task.id != curTaskId:
- continue
- return {
- "id": taskId,
- "projId": realProjId,
- "name": taskSeries.name,
- "url": fix_url(taskSeries.url),
- "locationId": taskSeries.location_id,
- "due": task.due,
"isCompleted": task.completed,
"completedDate": task.completed,
"priority": task.priority,
"estimate": task.estimate,
- "notes": list(self._get_notes(taskSeries.notes)),
+ "notes": dict((
+ (note["id"], note)
+ for note in self._get_notes(taskId, taskSeries.notes)
+ )),
}
+ taskDetails = self._parse_task_details(rawTaskDetails)
+ yield taskDetails
+
+ def get_task_details(self, taskId):
+ projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
+ for task in self.get_tasks_with_details(projId):
+ if task["id"] == taskId:
+ return task
return {}
def add_task(self, projId, taskName):
assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
def set_duedate(self, taskId, dueDate):
+ assert isinstance(dueDate, toolbox.Optional), (
+ "Date being set too definitively: %r" % dueDate
+ )
+
projId, seriesId, taskId = self._unpack_ids(taskId)
rsp = self._rtm.tasks.setDueDate(
timeline=self._timeline,
assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
def set_priority(self, taskId, priority):
- if priority is None:
- priority = "N"
- else:
- priority = str(priority)
+ assert isinstance(priority, toolbox.Optional), (
+ "Priority being set too definitively: %r" % priority
+ )
+ priority = str(priority.get_nothrow("N"))
projId, seriesId, taskId = self._unpack_ids(taskId)
rsp = self._rtm.tasks.setPriority(
)
assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
+ def add_note(self, taskId, noteTitle, noteBody):
+ projId, seriesId, taskId = self._unpack_ids(taskId)
+
+ rsp = self._rtm.tasks.notes.add(
+ timeline=self._timeline,
+ list_id=projId,
+ taskseries_id=seriesId,
+ task_id=taskId,
+ note_title=noteTitle,
+ note_text=noteBody,
+ )
+ assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
+
+ def update_note(self, noteId, noteTitle, noteBody):
+ projId, seriesId, taskId, note = self._unpack_ids(noteId)
+
+ rsp = self._rtm.tasks.notes.edit(
+ timeline=self._timeline,
+ note_id=noteId,
+ note_title=noteTitle,
+ note_text=noteBody,
+ )
+ assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
+
+ def delete_note(self, noteId):
+ projId, seriesId, taskId, noteId = self._unpack_ids(noteId)
+
+ rsp = self._rtm.tasks.notes.delete(
+ timeline=self._timeline,
+ note_id=noteId,
+ )
+ assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
+
@staticmethod
def _pack_ids(*ids):
"""
for task in tasks:
yield task
- def _get_notes(self, notes):
+ def _parse_task_details(self, rawTaskDetails):
+ taskDetails = {}
+ taskDetails["id"] = rawTaskDetails["id"]
+ taskDetails["projId"] = rawTaskDetails["projId"]
+ taskDetails["name"] = rawTaskDetails["name"]
+ taskDetails["url"] = fix_url(rawTaskDetails["url"])
+
+ rawLocationId = rawTaskDetails["locationId"]
+ if rawLocationId:
+ locationId = toolbox.Optional(rawLocationId)
+ else:
+ locationId = toolbox.Optional()
+ taskDetails["locationId"] = locationId
+
+ if rawTaskDetails["dueDate"]:
+ dueDate = datetime.datetime.strptime(
+ rawTaskDetails["dueDate"],
+ "%Y-%m-%dT%H:%M:%SZ",
+ )
+ dueDate = toolbox.Optional(dueDate)
+ else:
+ dueDate = toolbox.Optional()
+ taskDetails["dueDate"] = dueDate
+
+ taskDetails["isCompleted"] = len(rawTaskDetails["isCompleted"]) != 0
+
+ if rawTaskDetails["completedDate"]:
+ completedDate = datetime.datetime.strptime(
+ rawTaskDetails["completedDate"],
+ "%Y-%m-%dT%H:%M:%SZ",
+ )
+ completedDate = toolbox.Optional(completedDate)
+ else:
+ completedDate = toolbox.Optional()
+ taskDetails["completedDate"] = completedDate
+
+ try:
+ priority = toolbox.Optional(int(rawTaskDetails["priority"]))
+ except ValueError:
+ priority = toolbox.Optional()
+ taskDetails["priority"] = priority
+
+ if rawTaskDetails["estimate"]:
+ estimate = rawTaskDetails["estimate"]
+ estimate = toolbox.Optional(estimate)
+ else:
+ estimate = toolbox.Optional()
+ taskDetails["estimate"] = estimate
+
+ taskDetails["notes"] = rawTaskDetails["notes"]
+
+ rawKeys = list(rawTaskDetails.iterkeys())
+ rawKeys.sort()
+ parsedKeys = list(taskDetails.iterkeys())
+ parsedKeys.sort()
+ assert rawKeys == parsedKeys, "Missing some, %r != %r" % (rawKeys, parsedKeys)
+
+ return taskDetails
+
+ def _get_notes(self, taskId, notes):
if not notes:
return
elif isinstance(notes.note, list):
else:
notes = (notes.note, )
+ projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
+
for note in notes:
- id = note.id
+ noteId = self._pack_ids(projId, rtmSeriesId, rtmTaskId, note.id)
title = note.title
body = getattr(note, "$t")
yield {
- "id": id,
+ "id": noteId,
"title": title,
"body": body,
}