Created
August 25, 2010 16:52
-
-
Save zentrope/549843 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
from google.appengine.ext import webapp | |
from google.appengine.ext.webapp import util | |
from google.appengine.ext.webapp import template | |
from google.appengine.ext.webapp import WSGIApplication | |
from google.appengine.api import memcache | |
from google.appengine.ext import db | |
import logging | |
# =========================================================================== | |
# MODEL | |
# =========================================================================== | |
class User(db.Model): | |
name = db.StringProperty(required=True) | |
password = db.StringProperty(required=True) | |
email = db.StringProperty(required=True) | |
is_admin = db.BooleanProperty(default=False) | |
is_active = db.BooleanProperty(default=True) | |
@staticmethod | |
def get_by_uid(user): | |
query = User.gql('where name = :1', user) | |
return query.get() | |
@staticmethod | |
def get_user(user, passwd): | |
query = User.gql('where name = :1 and password = :2', user, passwd) | |
return query.get() | |
class Project(db.Model): | |
name = db.StringProperty(required=True) | |
description = db.TextProperty(required=True) | |
date_updated = db.DateTimeProperty(auto_now=True) | |
date_created = db.DateTimeProperty(auto_now_add=True) | |
owner = db.ReferenceProperty(User) | |
tasks = db.IntegerProperty(default=0) | |
completed_tasks = db.IntegerProperty(default=0) | |
def open_tasks(self): | |
if self.tasks == 0: | |
return 0 | |
return self.tasks - self.completed_tasks | |
class Task(db.Model): | |
project = db.ReferenceProperty(Project) | |
owner = db.ReferenceProperty(User) | |
summary = db.StringProperty(required=True) | |
notes = db.TextProperty(required=False) | |
status = db.StringProperty( | |
required=True, | |
default='open', | |
choices=['open', 'complete'] | |
) | |
date_updated = db.DateTimeProperty(auto_now=True) | |
date_created = db.DateTimeProperty(auto_now_add=True) | |
@staticmethod | |
def status_list(): | |
return ['open', 'complete'] | |
# =========================================================================== | |
class RootHandler(webapp.RequestHandler): | |
# This class wraps the standard RequestHandler so that we can | |
# add some handy methods likely to be needed by a lot of other | |
# handlers, such as session management and "canned" responses. | |
# The session id (really the user's identifier) is stored | |
# in a cookie. | |
cookie_name = 'taskbook' | |
def get_attr(self, name): | |
# Return a utf-8 encoded request attribute value, | |
# or an empty string if it's not present. | |
value = self.request.get(name) | |
if value is None: | |
return unicode('', 'utf-8') | |
# Return the value properly utf-8 encoded. | |
try: | |
return unicode(value.strip(), 'utf-8') | |
# If the value is already encoded, just return it. | |
except TypeError, e: | |
return value.strip() | |
def get_session_id(self): | |
# Return the ID stored in the session cookie, or None | |
# if it's not found. | |
self.request.charset = None | |
try: | |
return self.request.cookies[self.cookie_name] | |
except KeyError: | |
return None | |
def get_session_user(self): | |
# Return the user object for the currently logged in user | |
# or raise an exception if the user's not logged in. | |
uid = self.is_logged_in() | |
if not uid: | |
raise Exception, "Can't get user." | |
return User.get_by_uid(uid) | |
def is_logged_in(self): | |
# Tests whether or not there's an identifiable | |
# session id. | |
return self.get_session_id() | |
def sign_in(self, id): | |
# Establishes a session (kind of) by setting a cookie | |
# we can expect to see back on the next request. | |
cookie_str = "%s=%s;path=/;" % (self.cookie_name, id) | |
self.response.headers['set-cookie'] = cookie_str | |
def sign_out(self): | |
# Removes the session cookie so that the next request from | |
# the browser will seem like a branch new request. | |
the_past = 'Fri, 31-Dec-2000 23:59:59 GMT' | |
cookie_str = '%s=;expires=%s;path=/' % (self.cookie_name, the_past) | |
self.response.headers['set-cookie'] = cookie_str | |
def url_param(self, pattern): | |
# Given a pattern like /project/${id} and a path like /project/2, | |
# this method will return "2". In other words, its a way to extract | |
# REST style parameters from a URL path. Returns None if something | |
# goes wrong. | |
pt = pattern.split('/') | |
pa = self.request.path.split('/') | |
for x in range(0,len(pt)): | |
if pt[x].startswith('$'): | |
try: | |
return pa[x] | |
except IndexError: | |
return none | |
# =========================================================================== | |
def login_required(fn): | |
# Decorate handler methods with this if you want users to be | |
# redirected to a login page if they're not logged in, then returned | |
# to the page they requested when they've successfully logged in. | |
def new_fn(handler, *args, **kwargs): | |
uid = handler.is_logged_in() | |
if not uid: | |
data = { 'redirect_to' : handler.request.path } | |
page = template.render('www/html/login.html', data) | |
handler.response.out.write(page) | |
else: | |
return fn(handler, *args, **kwargs) | |
return new_fn | |
def user_required(fn): | |
# Makes sure that the user has logged in before invoking the decorated | |
# handler, or returns a 401 response. This should decorate handlers | |
# which are APIs (for Ajax calls, say) rather than handlers tasked | |
# with dispatching pages. | |
def new_fn(handler, *args, **kwargs): | |
uid = handler.is_logged_in() | |
if not uid: | |
handler.response.set_status(401) | |
return | |
else: | |
return fn(handler, *args, **kwargs) | |
return new_fn | |
# =========================================================================== | |
# Handlers meant to service Ajax API calls rather than to render pages | |
# of some sort. | |
class SignInApiHandler(RootHandler): | |
def post(self): | |
name = self.get_attr('name') | |
passwd = self.get_attr('password') | |
user = User.get_user(name, passwd) | |
if user: | |
logging.info('log in successful') | |
self.sign_in(name) | |
self.response.set_status(200) | |
else: | |
logging.info('log in unsuccessful') | |
self.response.set_status(401) | |
class TaskSummaryApiHandler(RootHandler): | |
@user_required | |
def post(self): | |
try: | |
task_id = self.url_param('/api/task/${id}/summary') | |
summary = self.get_attr('summary') | |
task = Task.get_by_id(int(task_id)) | |
user = self.get_session_user() | |
if user.key() != task.owner.key(): | |
logging.error('%s tried to update task note on task owned by %s' | |
% (user.name, task.owner.name)) | |
self.response.set_status(401) | |
return | |
task.summary = summary | |
task.put() | |
self.response.set_status(200) | |
return | |
except Exception, e: | |
logging.error("problem updating task summary: %s" % e) | |
self.response.set_status(500) | |
return | |
class TaskNoteApiHandler(RootHandler): | |
@user_required | |
def post(self): | |
try: | |
task_id = self.url_param('/api/task/${id}/note') | |
new_note = self.get_attr('note') | |
task = Task.get_by_id(int(task_id)) | |
user = self.get_session_user() | |
if user.key() != task.owner.key(): | |
logging.error('%s tried to update task note on task owned by %s' | |
% (user.name, task.owner.name)) | |
self.response.set_status(401) | |
return | |
task.notes = new_note | |
task.put() | |
self.response.set_status(200) | |
return | |
except Exception, e: | |
logging.error("problem updating task note: %s" % e) | |
self.response.set_status(500) | |
return | |
class TaskStatusApiHandler(RootHandler): | |
@user_required | |
def post(self): | |
try: | |
task_id = self.url_param('/api/task/${id}/status') | |
new_status = self.get_attr('status') | |
task = Task.get_by_id(int(task_id)) | |
user = self.get_session_user() | |
if user.key() != task.owner.key(): | |
logging.error('%s tried to update task status on task owned by %s' | |
% (user.name, task.owner.name)) | |
self.response.set_status(401) | |
return | |
task.status = new_status | |
task.put() | |
project = task.project | |
if new_status == 'open': | |
project.completed_tasks = project.completed_tasks - 1 | |
else: | |
project.completed_tasks = project.completed_tasks + 1 | |
project.put() | |
self.response.set_status(200) | |
return | |
except Exception, e: | |
logging.error("problem updating task status: %s" % e) | |
self.response.set_status(500) | |
return | |
class CreateTaskApiHandler(RootHandler): | |
@user_required | |
def post(self): | |
try: | |
uid = self.is_logged_in() | |
project_id = self.get_attr('project-id') | |
summary = self.get_attr('summary') | |
notes = self.get_attr('notes') | |
user = User.get_by_uid(uid) | |
project = Project.get_by_id(int(project_id)) | |
if project.owner.key() != user.key(): | |
# Note: this means an admin can't create a task on someone | |
# else's project either. I'm okay with that. | |
logging.error('%s tried to create a task on project owned by %s' | |
% (user.name, project.owner.name)) | |
self.response.set_status(401) | |
return | |
task = Task( | |
owner=user, | |
summary=summary, | |
project=project, | |
notes=notes | |
) | |
task.put() | |
project.tasks = project.tasks + 1 | |
project.put() | |
self.response.set_status(200) | |
return | |
except Exception, e: | |
logging.error("problem creating task: %s" % e) | |
self.response.set_status(500) | |
return | |
class CreateProjectApiHandler(RootHandler): | |
# TODO: Make sure that the project name is not already used for a given | |
# owner. | |
# TODO: Figure out a way to return an actual error message. Header? | |
@user_required | |
def post(self): | |
try: | |
uid = self.is_logged_in() | |
user = User.get_by_uid(uid) | |
name = self.get_attr('name') | |
description = self.get_attr('description') | |
project = Project(name=name, description=description, owner=user) | |
key = project.put() | |
self.response.set_status(200) | |
return | |
except Exception, e: | |
logging.error("problem creating project: %s" % e) | |
self.response.set_status(500) | |
return | |
# =========================================================================== | |
# Handlers for rendering pages | |
class DashboardPageHandler(RootHandler): | |
@login_required | |
def get(self): | |
user = self.get_session_user() | |
projects = Project.all() | |
projects.order('date_created') | |
if not user.is_admin: | |
projects.filter('owner =', user) | |
if projects.count() == 0: | |
projects = [] | |
data = { | |
'user' : user, | |
'projects' : projects | |
} | |
page = template.render('www/html/dashboard.html', data) | |
self.response.out.write(page) | |
class ProjectPageHandler(RootHandler): | |
@login_required | |
def get(self): | |
# get the user | |
user = self.get_session_user() | |
# get the project | |
p_id = self.url_param('/project/${id}') | |
if p_id is None: | |
self.response.set_status(404) | |
return | |
project = Project.get_by_id(int(p_id)) | |
if project is None: | |
self.response.set_status(404) | |
return | |
if not user.is_admin and project.owner.key() != user.key(): | |
self.response.set_status(401) | |
return | |
# get the tasks | |
tasks = Task.all() | |
tasks.filter('project = ', project) | |
tasks.order('date_created') | |
if tasks.count() == 0: | |
tasks = [] | |
# render the template | |
data = { 'user' : user, 'project' : project, 'tasks' : tasks } | |
page = template.render('www/html/project.html', data) | |
self.response.out.write(page) | |
class AboutHandler(RootHandler): | |
@login_required | |
def get(self): | |
user = self.get_session_user() | |
data = { 'user' : user } | |
page = template.render('www/html/about.html', data) | |
self.response.out.write(page) | |
class LogoutHandler(RootHandler): | |
def get(self): | |
self.sign_out() | |
self.redirect('/') | |
class LoginHandler(RootHandler): | |
def get(self): | |
if self.is_logged_in(): | |
self.redirect('/dashboard') | |
return | |
data = { | |
'redirect_to' : '/dashboard' | |
} | |
page = template.render('www/html/login.html', data) | |
self.response.out.write(page) | |
# =========================================================================== | |
# MAIN | |
# =========================================================================== | |
def init_user(name, passwd, email, admin=False): | |
user = User.get_user(name, passwd) | |
if user is None: | |
user = User(name=name, password=passwd, email=email) | |
user.is_admin = admin | |
user.put() | |
def init_users(): | |
init_user('admin', '-----', '[email protected]', admin=True) | |
init_user('keith', '-----', '[email protected]', admin=False) | |
init_user('demo', 'demo', '[email protected]', admin=False) | |
def main(): | |
handlers = [ | |
# callbacks for clients | |
('/api/sign-in', SignInApiHandler), | |
('/api/project', CreateProjectApiHandler), | |
('/api/task', CreateTaskApiHandler), | |
('/api/task/.*/status', TaskStatusApiHandler), | |
('/api/task/.*/note', TaskNoteApiHandler), | |
('/api/task/.*/summary', TaskSummaryApiHandler), | |
# actual rendered pages | |
('/logout', LogoutHandler), | |
('/login', LoginHandler), | |
('/about', AboutHandler), | |
('/dashboard', DashboardPageHandler), | |
('/project/.*', ProjectPageHandler), | |
# If all else fails, go to the log in page. TODO: dispatch | |
# to some friendly version of a 404 page letting the user know | |
# why it is they're not seeing something they expect to see. | |
('/.*', LoginHandler) | |
] | |
application = webapp.WSGIApplication(handlers, debug=True) | |
util.run_wsgi_app(application) | |
if __name__ == '__main__': | |
init_users() | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
This is the python part of a "task manager" application implemented on top of Google App Engine. Most of the fun stuff is in the jQuery stuff on the client.
You can play around with the app at: http://kfi-taskbook.appspot.com, login: demo/demo.
I'm not sure if the app works well on non-Apple versions of various browsers, mainly because of the key bindings. For instance, "alt-click" to edit a task, or the note attached to a task, is a bit challenging, even with jQuery to help you out.