mygpo/db/couchdb/__init__.py

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