A Messy Demo

async-demo.py

#!/usr/bin/env python
# encoding: utf-8
"""
async-demo.py

Created by Christian Swinehart on 2012-02-14.
Copyright (c) 2012 Samizdat Drafting Co. All rights reserved.
"""

from __future__ import with_statement
import sys
import os
import random

# dependencies: pip install tornado decorator
from decorator import decorator
from couchdb import Server, Database, ResourceConflict
from tornado import ioloop
import tornado.web

def pretty(count, name):
    return "%i %s%s:"%(count, name, '' if count==1 else 's')


def basic_crud():
    # grab these synchronously so there aren't sequence problems below
    s = Server('http://127.0.0.1:5984')
    db = s['cc']

    # get a database (after seeing it exists) without blocking
    def got_db(database, error):
        if error: 
            print "Server.get: Error", error
        else:
            print "Database:", database
    s.get('cc', callback=got_db)
    
    # get a single doc
    def got_doc(doc, error):
        if error: 
            print "Database.get: Error", error
        print "Document:", doc
    db.get('lorem', callback=got_doc)
        
    # get a selection of docs by id
    def got_many(docs, error):
        if error: 
            print "Database.get: Error", error
        print pretty(len(docs),"Document")
        for doc in docs:
            print "  %s"%doc
    db.get(['lorem','ipsum','dolor'], callback=got_many)
    
    # access a view
    def got_rows(rows, error):
        if error: 
            print "Database.view: Error", error
        print pretty(len(rows), "Row")
        for row in rows:
            print "  %s"%row
    db.view('_all_docs', limit=20, descending=True, callback=got_rows)


def changes_feed():
    s = Server('http://127.0.0.1:5984')
    db = s['cc']
    
    def callback(seq, changes):
        # the callback will be continually invoked whenever db writes occur.
        # calls are throttled to a maximum rate of 1/s since edits often come
        # in barrages
        print "Current seq:",seq
        print pretty(len(changes),"Change"), changes, "\n"
        
    db.changes(feed="continuous", heartbeat=60000, since=11430, callback=callback)
    

def document_readwrite():
    # grab these synchronously so there aren't sequence problems below
    s = Server('http://127.0.0.1:5984')
    db = s['cc']
    some_docs = [row.doc for row in db.view('_all_docs', include_docs=True, limit=5)]

    # make some edits to the documents we loaded earlier and PUT them to the db
    for doc in some_docs:
        doc['foo'] = random.random()
        # randomly delete the revs on a few docs to cause a conflict when we upload
        if random.random()<.1: 
            del doc['_rev']
        
    def done_writing(resp, error):
        if error: raise error
        # the resp is a list of {id:, doc:, error:} dicts:
        
        # find the docs that weren't updated due to stale/missing _revs
        conflicts = [r['doc'] for r in resp if isinstance(r['error'], ResourceConflict)]
        if conflicts:
            print "Found conflics when writing:",conflicts

            # fetch the current versions of the conflicted docs so we can merge them
            def got_old_versions(docs, error):
                # let's just overwrite whatever was on the database by copying the 
                # current _rev onto the doc we already tried to upload. Then we can 
                # retry the write.
                merged = []
                for old_doc, new_doc in zip(conflicts, docs):
                    if new_doc:
                        old_doc['_rev'] = new_doc['_rev']
                        merged.append(old_doc)

                def wrote_merged_versions(resp, error):
                    # hopefully the rev-transplant and reupload was quick enough
                    # that nobody else edited the doc in the meantime and we won't
                    # have any more conflicts
                    if error: raise error
                    
                    if any(r for r in resp if r['error']):
                        print "Still have conflicts:", resp
                    else:
                        print "Successfully merged"
                    
                    
                db.update(merged, callback=wrote_merged_versions)
                
            db.get(conflicts, callback=got_old_versions)

    db.update(some_docs, callback=done_writing)


def webapp():
    db = Database('http://127.0.0.1/db')
    
    class HelloYou(tornado.web.RequestHandler):    
        @tornado.web.asynchronous
        def get(self, user_id):
            db.get(user_id, callback=self.got_user_doc)
            
        def got_user_doc(self, doc, error):
            # if error isn't None, something went wrong
            if error: 
                self.write('hello whoever you are')
                print error
            else:
                # presume the doc is of the form {firstname:"", lastname:""}
                self.write('hello %s %s'%(doc['firstname'],doc['lastname']))
            self.finish()
            
    
    application = tornado.web.Application([
      (r"/hello/([^/]+)", HelloYou),
    ])
    application.listen(1920)
    loop = tornado.ioloop.IOLoop.instance()
    loop.start()

    
def main():
    # the main example
    webapp()
    
    # try running these one at a time ... and then all at once...
    # basic_crud()
    # document_readwrite()
    # changes_feed()    
    
    # this coordinates the http and keeps the script running until the loop
    # is stopped in one of the callbacks (or you hit ^c). otherwise it would
    # exit almost immediately.
    # ioloop.IOLoop.instance().start() 





if __name__ == '__main__':
    main()