Navigation works!
This commit is contained in:
3
Makefile
3
Makefile
@@ -1,3 +1,6 @@
|
||||
all:
|
||||
git submodule init
|
||||
git submodule update
|
||||
|
||||
tags: ui/*.py main.py
|
||||
ctags -R ui main.py
|
||||
|
||||
66
asana_service.py
Normal file
66
asana_service.py
Normal file
@@ -0,0 +1,66 @@
|
||||
from models.models import *
|
||||
|
||||
class AsanaService(object):
|
||||
TASK_FIELDS = [
|
||||
'name',
|
||||
'notes',
|
||||
'assignee.name',
|
||||
'assignee_status',
|
||||
'completed',
|
||||
'due_on',
|
||||
'due_at',
|
||||
'projects.name',
|
||||
'parent.completed',
|
||||
'parent.name',
|
||||
'memberships.section.name',
|
||||
'memberships.project.name',
|
||||
'custom_fields.name',
|
||||
'custom_fields.type',
|
||||
'custom_fields.text_value',
|
||||
'custom_fields.number_value',
|
||||
'custom_fields.enum_value.name',
|
||||
]
|
||||
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
self.completed_tasks = False
|
||||
self.me = User(client.users.me())
|
||||
self.workspace = self.me.workspaces()[0]
|
||||
|
||||
def __wrap__(self, Klass, values):
|
||||
return map(Klass, values)
|
||||
|
||||
def get_tasks(self, project_id):
|
||||
params = {
|
||||
'completed_since': '' if self.completed_tasks else 'now',
|
||||
'opt_fields': self.TASK_FIELDS,
|
||||
}
|
||||
|
||||
return self.__wrap__(
|
||||
Task,
|
||||
self.client.tasks.find_by_project(project_id, params=params)
|
||||
)
|
||||
|
||||
def get_my_tasks(self):
|
||||
return self.__wrap__(
|
||||
Task,
|
||||
self.client.tasks.find_all(params = {
|
||||
'completed_since': '' if self.completed_tasks else 'now',
|
||||
'opt_fields': self.TASK_FIELDS,
|
||||
'assignee': self.me.id(),
|
||||
'workspace': self.workspace.id()
|
||||
})
|
||||
)
|
||||
|
||||
def get_task(self, task_id):
|
||||
return Task(self.client.tasks.find_by_id(
|
||||
task_id,
|
||||
params={
|
||||
'opt_fields': self.TASK_FIELDS
|
||||
}
|
||||
))
|
||||
|
||||
def get_stories(self, task_id):
|
||||
return self.__wrap__(Story,
|
||||
self.client.stories.find_by_task(task_id)
|
||||
)
|
||||
43
main.py
43
main.py
@@ -9,16 +9,18 @@ import urwid
|
||||
|
||||
import secrets
|
||||
from ui.auth import AuthPrompt
|
||||
from ui.palette import palette
|
||||
from ui.constants import palette
|
||||
from ui.ui import Ui, loading
|
||||
|
||||
from asana_service import AsanaService
|
||||
|
||||
class NotAuthedException(Exception):
|
||||
def __init__(self):
|
||||
super(NotAuthedException, self)
|
||||
|
||||
class CmdAsana(object):
|
||||
"""The main urwid loop
|
||||
"""
|
||||
loop = None
|
||||
nav_stack = []
|
||||
|
||||
"""Try to get an Asana client using stored tokens
|
||||
|
||||
@@ -37,6 +39,10 @@ class CmdAsana(object):
|
||||
token=token,
|
||||
token_updater=self.save_token,
|
||||
auto_refresh_url=AsanaOAuth2Session.token_url,
|
||||
auto_refresh_kwargs={
|
||||
'client_id': secrets.CLIENT_ID,
|
||||
'client_secret': secrets.CLIENT_SECRET
|
||||
}
|
||||
)
|
||||
except IOError:
|
||||
raise NotAuthedException()
|
||||
@@ -48,6 +54,10 @@ class CmdAsana(object):
|
||||
redirect_uri='urn:ietf:wg:oauth:2.0:oob',
|
||||
token_updater=self.save_token,
|
||||
auto_refresh_url=AsanaOAuth2Session.token_url,
|
||||
auto_refresh_kwargs={
|
||||
'client_id': secrets.CLIENT_ID,
|
||||
'client_secret': secrets.CLIENT_SECRET
|
||||
}
|
||||
)
|
||||
(url, _) = self.client.session.authorization_url()
|
||||
auth = AuthPrompt(url, self.auth_callback)
|
||||
@@ -74,9 +84,31 @@ class CmdAsana(object):
|
||||
def exit_handler(self, key):
|
||||
if key in ('q', 'Q', 'esc'):
|
||||
raise urwid.ExitMainLoop()
|
||||
print(key)
|
||||
if key in ('backspace'):
|
||||
self.ui.go_back()
|
||||
|
||||
def get_asana_service(self):
|
||||
self.asana_service = AsanaService(self.client)
|
||||
|
||||
def get_ui(self):
|
||||
self.ui = Ui(self.asana_service, self.update)
|
||||
|
||||
def run(self):
|
||||
print("Running...", self.client.users.me())
|
||||
self.placeholder = urwid.WidgetPlaceholder(loading())
|
||||
self.loop = urwid.MainLoop(
|
||||
self.placeholder,
|
||||
unhandled_input=self.exit_handler,
|
||||
palette=palette
|
||||
)
|
||||
self.ui.my_tasks()
|
||||
|
||||
def update(self, widget):
|
||||
self.loop.widget.original_widget = widget
|
||||
try:
|
||||
self.loop.draw_screen()
|
||||
except Exception:
|
||||
self.loop.run()
|
||||
|
||||
def save_token(self, token):
|
||||
f = open('.oauth', 'w')
|
||||
@@ -91,6 +123,9 @@ def main():
|
||||
except NotAuthedException:
|
||||
cmd_asana.authorize()
|
||||
|
||||
cmd_asana.get_asana_service()
|
||||
cmd_asana.get_ui()
|
||||
|
||||
cmd_asana.run()
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
0
models/__init__.py
Normal file
0
models/__init__.py
Normal file
93
models/models.py
Normal file
93
models/models.py
Normal file
@@ -0,0 +1,93 @@
|
||||
import dateutil
|
||||
import sys
|
||||
|
||||
class AsanaObject(object):
|
||||
def __init__(self, object_dict):
|
||||
self.object_dict = object_dict
|
||||
|
||||
def id(self):
|
||||
return self.object_dict['id']
|
||||
|
||||
def name(self):
|
||||
if 'name' in self.object_dict:
|
||||
return self.object_dict['name']
|
||||
else:
|
||||
return ''
|
||||
|
||||
class User(AsanaObject):
|
||||
def workspaces(self):
|
||||
return [AsanaObject(w) for w in self.object_dict['workspaces']]
|
||||
|
||||
class Task(AsanaObject):
|
||||
def name(self):
|
||||
if self.object_dict['completed']:
|
||||
return '✓ %s' % super(self).name()
|
||||
return super(Task, self).name()
|
||||
|
||||
def assignee(self):
|
||||
if 'assignee' in self.object_dict:
|
||||
return User(self.object_dict['assignee'])
|
||||
else:
|
||||
return None
|
||||
|
||||
def atm_section(self):
|
||||
return self.object_dict['assignee_status']
|
||||
|
||||
def description(self):
|
||||
if 'notes' in self.object_dict:
|
||||
return self.object_dict['notes']
|
||||
else:
|
||||
return None
|
||||
|
||||
def due_date(self):
|
||||
if 'due_at' in self.object_dict:
|
||||
return dateutil.parser.parse(self.object_dict['due_at'])
|
||||
elif 'due_one' in self.object_dict:
|
||||
return dateutil.parser.parse(self.object_dict['due_on'])
|
||||
else:
|
||||
return None
|
||||
|
||||
def parent(self):
|
||||
if 'parent' in self.object_dict and self.object_dict['parent']:
|
||||
return Task(self.object_dict['parent'])
|
||||
else:
|
||||
return None
|
||||
|
||||
def projects(self):
|
||||
if 'projects' in self.object_dict:
|
||||
return [Project(p) for p in self.object_dict['projects']]
|
||||
else:
|
||||
return []
|
||||
|
||||
def custom_fields(self):
|
||||
if 'custom_fields' in self.object_dict:
|
||||
return [CustomField(c) for c in self.object_dict['custom_fields']]
|
||||
else:
|
||||
return []
|
||||
class Project(AsanaObject):
|
||||
def description(self):
|
||||
if 'notes' in self.object_dict:
|
||||
return self.object_dict['notes']
|
||||
else:
|
||||
return ''
|
||||
|
||||
class CustomField(AsanaObject):
|
||||
def string_value(self):
|
||||
if 'text_value' in self.object_dict:
|
||||
return self.object_dict['text_value']
|
||||
elif 'number_value' in self.object_dict:
|
||||
return self.object_dict['number_value']
|
||||
elif 'enum_value' in self.object_dict and self.object_dict['enum_value']:
|
||||
enum_value = AsanaObject(self.object_dict['enum_value'])
|
||||
return enum_value.name()
|
||||
|
||||
return ''
|
||||
|
||||
class Story(AsanaObject):
|
||||
def string_value(self):
|
||||
if 'created_by' in self.object_dict:
|
||||
creator = self.object_dict['created_by']['name'] + ' '
|
||||
else:
|
||||
creator = ''
|
||||
|
||||
return '%s%s' % (creator, self.object_dict['text'])
|
||||
15
ui/auth.py
15
ui/auth.py
@@ -1,9 +1,12 @@
|
||||
import urwid
|
||||
|
||||
"""
|
||||
Input box that accepts OAuth tokens
|
||||
"""
|
||||
class TokenEdit(urwid.Edit):
|
||||
def __init__(self):
|
||||
urwid.register_signal(TokenEdit, 'TokenEdit-changed')
|
||||
prompt = ('seondary', u'Auth Token: ')
|
||||
prompt = ('seondary', u' Authorization Token: ')
|
||||
super(TokenEdit, self).__init__(prompt, '')
|
||||
|
||||
def keypress(self, size, key):
|
||||
@@ -18,12 +21,16 @@ class AuthPrompt(object):
|
||||
token_input = TokenEdit()
|
||||
urwid.connect_signal(token_input, 'TokenEdit-changed', self.callback)
|
||||
|
||||
self.frame = urwid.Filler(
|
||||
self.frame = urwid.Filler(urwid.Padding(
|
||||
urwid.Pile([
|
||||
urwid.Text('Visit %s and paste the token below.\n' % auth_url),
|
||||
token_input,
|
||||
])
|
||||
)
|
||||
]),
|
||||
align='center',
|
||||
width='pack',
|
||||
left=2,
|
||||
right=2
|
||||
))
|
||||
|
||||
def callback(self, token):
|
||||
self.callback(token)
|
||||
|
||||
@@ -2,10 +2,15 @@ palette = [
|
||||
('selected', 'standout', ''),
|
||||
('selected workspace', 'standout,bold', ''),
|
||||
('header', 'bold,light green', ''),
|
||||
('secondary', 'light gray', ''),
|
||||
('secondary', 'light green', ''),
|
||||
('task', 'light green', ''),
|
||||
('project', 'yellow', ''),
|
||||
('section', 'white', 'dark green'),
|
||||
('section', 'dark green,bold', ''),
|
||||
('atm_section', 'white,bold', 'dark blue'),
|
||||
('workspace', 'white', 'dark blue'),
|
||||
('pager', 'standout', ''),
|
||||
]
|
||||
|
||||
keys = {
|
||||
'select': ['enter', 'space']
|
||||
}
|
||||
62
ui/task_details.py
Normal file
62
ui/task_details.py
Normal file
@@ -0,0 +1,62 @@
|
||||
import urwid
|
||||
|
||||
class TaskDetails(object):
|
||||
def __init__(self, task, stories, on_subtask_click, on_project_click,
|
||||
on_comment):
|
||||
self.task = task
|
||||
self.on_subtask_click = on_subtask_click,
|
||||
self.on_project_click = on_project_click,
|
||||
self.on_comment = on_comment
|
||||
|
||||
self.details = urwid.Pile([
|
||||
('pack', urwid.Text(('task', task.name()))),
|
||||
('pack', urwid.Divider('-')),
|
||||
('weight', 1, Memberships(task, on_subtask_click, on_project_click) \
|
||||
.component()),
|
||||
('pack', urwid.Divider('-')),
|
||||
('pack', CustomFields(task).component()),
|
||||
('pack', urwid.Divider('-')),
|
||||
('weight', 20, urwid.Filler(urwid.Text(task.description()))),
|
||||
('weight', 5, urwid.Filler(Stories(stories).component()))
|
||||
])
|
||||
|
||||
def component(self):
|
||||
return self.details
|
||||
|
||||
class Memberships(object):
|
||||
def __init__(self, task, on_subtask_click, on_project_click):
|
||||
components = [urwid.Button(
|
||||
('project', p.name()),
|
||||
on_press = lambda x: on_project_click(p.id())
|
||||
) for p in task.projects()]
|
||||
if task.parent():
|
||||
components.append(urwid.Button(
|
||||
('task', 'Subtask of: %s' % task.parent().name()),
|
||||
on_press = lambda x: on_subtask_click(task.parent().id())
|
||||
))
|
||||
|
||||
self.memberships = urwid.ListBox(
|
||||
urwid.SimpleFocusListWalker(components)
|
||||
)
|
||||
|
||||
def component(self):
|
||||
return self.memberships
|
||||
|
||||
class CustomFields(object):
|
||||
def __init__(self, task):
|
||||
components = [urwid.Text('%s: %s' % (f.name(), f.string_value()))
|
||||
for f in task.custom_fields()]
|
||||
|
||||
self.custom_fields = urwid.Pile(components)
|
||||
|
||||
def component(self):
|
||||
return self.custom_fields
|
||||
|
||||
class Stories(object):
|
||||
def __init__(self, stories):
|
||||
components = [urwid.Text(s.string_value()) for s in stories]
|
||||
|
||||
self.stories = urwid.Pile(components)
|
||||
|
||||
def component(self):
|
||||
return self.stories
|
||||
63
ui/task_list.py
Normal file
63
ui/task_list.py
Normal file
@@ -0,0 +1,63 @@
|
||||
import urwid
|
||||
|
||||
from ui.constants import keys
|
||||
|
||||
class TaskList(object):
|
||||
def __init__(self, tasks, header, on_task_click):
|
||||
self.callback = on_task_click
|
||||
self.grid = urwid.Frame(
|
||||
urwid.ListBox(
|
||||
urwid.SimpleFocusListWalker(
|
||||
[TaskRow(t, self.on_task_clicked) for t in tasks]
|
||||
)
|
||||
),
|
||||
header=urwid.Text(header),
|
||||
focus_part='body'
|
||||
)
|
||||
|
||||
def on_task_clicked(self, id):
|
||||
self.callback(id)
|
||||
|
||||
def component(self):
|
||||
return self.grid
|
||||
|
||||
class MyTasks(object):
|
||||
def __init__(self, tasks, on_task_click):
|
||||
all_tasks = [t for t in tasks]
|
||||
|
||||
today = [t for t in all_tasks if t.atm_section() == 'today']
|
||||
upcoming = [t for t in all_tasks if t.atm_section() == 'upcoming']
|
||||
later = [t for t in all_tasks if t.atm_section() == 'later']
|
||||
|
||||
self.today_grid = TaskList(today,
|
||||
('atm_section', 'Today'),
|
||||
on_task_click)
|
||||
self.upcoming_grid = TaskList(upcoming,
|
||||
('atm_section', 'Upcoming'),
|
||||
on_task_click)
|
||||
self.later_grid = TaskList(later,
|
||||
('atm_section', 'Later'),
|
||||
on_task_click)
|
||||
|
||||
def component(self):
|
||||
return urwid.Frame(urwid.Pile([
|
||||
self.today_grid.component(),
|
||||
self.upcoming_grid.component(),
|
||||
self.later_grid.component()
|
||||
]),
|
||||
header=urwid.Text(('header', 'My Tasks')),
|
||||
focus_part='body'
|
||||
)
|
||||
|
||||
class TaskRow(urwid.SelectableIcon):
|
||||
def __init__(self, task, on_click):
|
||||
self.on_click = on_click
|
||||
self.task = task
|
||||
style = 'section' if task.name()[-1] == ':' else 'task'
|
||||
super(TaskRow, self).__init__((style, task.name()))
|
||||
|
||||
def keypress(self, size, key):
|
||||
if key in keys['select']:
|
||||
self.on_click(self.task.id())
|
||||
else:
|
||||
return key
|
||||
72
ui/ui.py
Normal file
72
ui/ui.py
Normal file
@@ -0,0 +1,72 @@
|
||||
import urwid
|
||||
from threading import Thread
|
||||
|
||||
from asana_service import AsanaService
|
||||
|
||||
from ui.task_list import MyTasks, TaskList
|
||||
from ui.task_details import TaskDetails
|
||||
|
||||
class Ui(object):
|
||||
nav_stack = []
|
||||
def __init__(self, asana_service, update):
|
||||
self.asana_service = asana_service
|
||||
self.update = update
|
||||
|
||||
def my_tasks(self):
|
||||
self.nav_stack.append(('mytasks', None))
|
||||
|
||||
def runInThread():
|
||||
tasks = self.asana_service.get_my_tasks()
|
||||
self.update(MyTasks(tasks, self.task_details).component())
|
||||
|
||||
thread = Thread(target=runInThread())
|
||||
thread.start()
|
||||
|
||||
|
||||
def task_details(self, id):
|
||||
self.nav_stack.append(('task', id))
|
||||
def runInThread():
|
||||
task = self.asana_service.get_task(id)
|
||||
stories = self.asana_service.get_stories(id)
|
||||
self.update(TaskDetails(task,
|
||||
stories,
|
||||
self.task_details,
|
||||
self.task_list,
|
||||
None).component())
|
||||
thread = Thread(target=runInThread())
|
||||
thread.start()
|
||||
|
||||
def task_list(self, id):
|
||||
self.nav_stack.append(('project', id))
|
||||
def runInThread():
|
||||
tasks = self.asana_service.get_tasks(id)
|
||||
self.update(TaskList(tasks,
|
||||
'TODO: get project name',
|
||||
self.task_details
|
||||
).component())
|
||||
thread = Thread(target=runInThread())
|
||||
thread.run()
|
||||
|
||||
def go_back(self):
|
||||
if len(self.nav_stack) < 2:
|
||||
return
|
||||
|
||||
self.nav_stack.pop()
|
||||
(location, id) = self.nav_stack.pop()
|
||||
if location == 'mytasks':
|
||||
self.my_tasks()
|
||||
elif location == 'task':
|
||||
self.task_details(id)
|
||||
elif location == 'project':
|
||||
self.task_list(id)
|
||||
|
||||
|
||||
def loading():
|
||||
return urwid.Overlay(
|
||||
urwid.BigText('Loading...', urwid.font.HalfBlock5x4Font()),
|
||||
urwid.SolidFill('#'),
|
||||
'center',
|
||||
'pack',
|
||||
'middle',
|
||||
'pack'
|
||||
)
|
||||
Reference in New Issue
Block a user