Source

avalanche / doc / tutorial / tasks / tasks.py

Full commit
import jinja2
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.expression import func
from sqlalchemy.orm import sessionmaker

from avalanche.core import RequestHandler
from avalanche.core import WSGIApplication
from avalanche.router import Route
from avalanche import make_handler, BaseHandler, RedirectTo
from avalanche.renderer import JinjaRenderer
from avalanche.params import UrlPathParam, UrlQueryParam, PostGroupParam
from avalanche.params import ContextParam


BaseModel = declarative_base()

class Task(BaseModel):
    __tablename__ = 'task'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    closed = Column(Boolean, default=False)
    created = Column(DateTime, server_default=func.now())



############ Request Handlers

class FlashMixinHandler(BaseHandler):
    @UrlQueryParam('flash')
    def ctx_flash(self, flash=None):
        return {'flash': flash}


class TaskRequestHandler(FlashMixinHandler, BaseHandler):

    MSG_TASK_NEW = 'New task successfully added'
    MSG_TASK_NAME = 'Please enter a name for the task'
    MSG_TASK_CLOSED = 'Task was successfully closed'

    JINJA_ENV = jinja2.Environment(
        loader=jinja2.FileSystemLoader('templates'),
        undefined=jinja2.DebugUndefined,
        autoescape=True,
        )

    renderer = JinjaRenderer(JINJA_ENV)

    context_get = ['db_session', 'a_get', 'ctx_flash',]
    context_post = ['db_session', 'a_post']

    def db_session(self):
        return {'db': self.app.Session()}



class ListTasks(TaskRequestHandler):
    """list all open tasks"""
    template = 'task_list.html'

    @ContextParam('db')
    @UrlQueryParam('show_closed', 'closed', bool)
    def a_get(self, db, show_closed=False):
        tasks = db.query(Task).order_by(Task.created)
        if not show_closed:
            tasks = tasks.filter(Task.closed==False)
        return {
            'tasks': tasks,
            'show_closed': show_closed,
            }


class NewTask(TaskRequestHandler):
    """add a new task"""
    template = 'task_new.html'

    def a_get(self): # pragma: no cover
        pass

    @ContextParam('db')
    @PostGroupParam('data', 'task')
    def a_post(self, db, data):
        name = data.get('name')
        if name:
            db.add(Task(name=name))
            db.commit()
            return RedirectTo('task-list', flash=self.MSG_TASK_NEW)
        else:
            return RedirectTo('task-new', flash=self.MSG_TASK_NAME)

class CloseTask(TaskRequestHandler):
    """mark a task as closed"""

    @ContextParam('db')
    @UrlPathParam('task_id')
    def a_post(self, db, task_id):
        task = db.query(Task).get(int(task_id))
        task.closed = True
        db.commit()
        return RedirectTo('task-list', flash=self.MSG_TASK_CLOSED)



########### WSGI App

def tasks_app(Session):
    route_raw = [('/', ListTasks, 'task-list'),
                 ('/new', NewTask, 'task-new'),
                 ('/close/<task_id>', CloseTask, 'task-close')
                 ]
    routes = []
    for path, handler, name in route_raw:
        handler_class = make_handler(RequestHandler, handler)
        routes.append(Route(path, handler_class, name))
    wsgi_app = WSGIApplication(routes, debug=True)
    wsgi_app.Session = Session
    return wsgi_app

if __name__ == '__main__': # pragma: no cover
    from wsgiref.simple_server import make_server
    engine = create_engine('sqlite:///tasks.db', echo=True)
    BaseModel.metadata.create_all(engine)
    app = tasks_app(Session=sessionmaker(bind=engine))
    server = make_server('0.0.0.0', 8080, app)
    print('Server started on port 8080\n\n')
    server.serve_forever()