mygpo/couchdb.py

   1 from operator import itemgetter
   2 from collections import namedtuple
   3
   4 from couchdbkit import *
   5
   6 class BulkException(Exception):
   7
   8     def __init__(self, errors):
   9         self.errors = errors
  10
  11
  12 BulkError = namedtuple('BulkError', 'doc error reason')
  13
  14
  15 def __default_reload(db, obj):
  16     _id = getattr(obj, '_id', obj.get('_id', None))
  17
  18     if isinstance(obj, Document):
  19         return obj.__class__.get(_id)
  20     else:
  21         return db[_id]
  22
  23
  24 __get_obj = itemgetter(0)
  25
  26 def bulk_save_retry(db, obj_funs, reload_f=__default_reload):
  27     """ Saves multiple documents and retries failed ones
  28
  29     Objects to be saved are passed as (obj, mod_f), where obj is the CouchDB
  30     and mod_f is the modification function that should be applied to it.
  31
  32     If saving a document fails, it is again fetched from the database, the
  33     modification function is applied again and saving is retried. """
  34
  35     errors = []
  36
  37     while True:
  38
  39         # apply modification function (and keep funs)
  40         obj_funs = [(f(o), f) for (o, f) in obj_funs]
  41
  42         # filter those with obj None
  43         obj_funs = filter(lambda of: __get_obj(of) is not None, obj_funs)
  44
  45         # extract objects
  46         objs = map(__get_obj, obj_funs)
  47
  48         if not objs:
  49             return
  50
  51         try:
  52             db.save_docs(objs)
  53             return
  54
  55         except BulkSaveError as ex:
  56
  57             new_obj_funs = []
  58             for res, (obj, f) in zip(ex.results, obj_funs):
  59                 if res.get('error', False) == 'conflict':
  60
  61                     # reload conflicted object
  62                     obj = reload_f(db, obj)
  63                     new_obj_funs.append( (obj, f) )
  64
  65                 elif res.get('error', False):
  66                     # don't retry other errors
  67                     err = BulkError(obj, res['error'], res.get('reason', None))
  68                     errors.append(err)
  69
  70             obj_funs = new_obj_funs
  71
  72     if errors:
  73         raise BulkException(errors)