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