Process files attached to couchdb docs

change_monitor.py

import traceback

import couchdb

couch = couchdb.Server('http://admin:admin@127.0.0.1:5984/')

try:
    db = couch.create('test1')
except Exception, e:
    traceback.print_exc()
    db = couch['test1']

ch = db.changes(feed='continuous', heartbeat='1000', include_docs=True)

for doc in ch:
    try:
        if '_attachments' in doc['doc']:
            for k, v in doc['doc']['_attachments'].iteritems():
                with open('{}-{}'.format(doc['id'], k), 'w') as fd:
                    response = db.get_attachment(doc['id'], k)
                    buffer_len = 16 * 1024
                    fd.write('{},{},'.format(doc['id'], doc['doc']['datetime']))
                    while True:
                        buf = response.read(buffer_len).strip()
                        if not buf:
                            break

                        line = '\n{},{},'.format(doc['id'], doc['doc']['datetime'])
                        buf = buf.replace('\n', line)
                        fd.write(buf)

    except KeyError, e:
        traceback.print_exc()
    except Exception, e:
        raise

producer.py

import StringIO
import datetime
import threading

import couchdb

couch = couchdb.Server('http://admin:admin@127.0.0.1:5984/')
try:
    db = couch.create('test1')
except:
    db = couch['test1']


def work():
    for k in range(10):
        doc = {'thread': threading.currentThread().getName(), 'count': k,
               'datetime': datetime.datetime.utcnow().isoformat()}
        db.save(doc)
        buf = StringIO.StringIO("")
        for j in range(1000):
            buf.write("{}, TEST ATTACHMENT FROM, {}\n".format(j, threading.currentThread().getName()))

        # print buf.getvalue()
        db.put_attachment(doc, buf.getvalue(), filename="sample.txt")
        buf.close()


for i in range(10):
    t = threading.Thread(target=work)
    t.start()