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