mygpo/db/couchdb/__init__.py

   1 from operator import itemgetter
   2 from collections import namedtuple
   3
   4 from couchdbkit.ext.django import loading
   5 from couchdbkit import MultipleResultsFound
   6 from couchdbkit import *
   7
   8 import logging
   9 logger = logging.getLogger(__name__)
  10
  11
  12 def get_userdata_database():
  13     return loading.get_db('userdata')
  14
  15
  16 class BulkException(Exception):
  17
  18     def __init__(self, errors):
  19         self.errors = errors
  20
  21
  22 BulkError = namedtuple('BulkError', 'doc error reason')
  23
  24
  25 def __default_reload(db, obj):
  26     doc = db[obj._id]
  27     return obj.__class__.wrap(doc)
  28
  29
  30 __get_obj = itemgetter(0)
  31
  32 def bulk_save_retry(obj_funs, db, reload_f=__default_reload):
  33     """ Saves multiple documents and retries failed ones
  34
  35     Objects to be saved are passed as (obj, mod_f), where obj is the CouchDB
  36     and mod_f is the modification function that should be applied to it.
  37
  38     If saving a document fails, it is again fetched from the database, the
  39     modification function is applied again and saving is retried. """
  40
  41     errors = []
  42
  43     while True:
  44
  45         # apply modification function (and keep funs)
  46         obj_funs = [(f(o), f) for (o, f) in obj_funs]
  47
  48         # filter those with obj None
  49         obj_funs = filter(lambda of: __get_obj(of) is not None, obj_funs)
  50
  51         # extract objects
  52         objs = map(__get_obj, obj_funs)
  53
  54         if not objs:
  55             return
  56
  57         try:
  58             db.save_docs(objs)
  59             return
  60
  61         except BulkSaveError as ex:
  62
  63             new_obj_funs = []
  64             for res, (obj, f) in zip(ex.results, obj_funs):
  65                 if res.get('error', False) == 'conflict':
  66
  67                     # reload conflicted object
  68                     obj = reload_f(db, obj)
  69                     new_obj_funs.append( (obj, f) )
  70
  71                 elif res.get('error', False):
  72                     # don't retry other errors
  73                     err = BulkError(obj, res['error'], res.get('reason', None))
  74                     errors.append(err)
  75
  76             obj_funs = new_obj_funs
  77
  78     if errors:
  79         logger.warn('Errors at bulk-save: %s', errors)
  80         raise BulkException(errors)
  81
  82
  83 def get_single_result(db, view, **query_args):
  84     """ return a single CouchDB view result
  85
  86     Logs an error if multiple results are returned, and uses the first result.
  87     This can happen as CouchDB can not guarantee uniqueness of attributes other
  88     than _id. If no result are fetched, None is returned. """
  89
  90     r = db.view(view, **query_args)
  91
  92     if not r:
  93         return None
  94
  95     try:
  96         result = r.one()
  97
  98     except MultipleResultsFound as ex:
  99         logger.info('Multiple results found in %s with params %s',
 100                     view, query_args)
 101         # use the first result as fallback
 102         result = r.first()
 103
 104     # we can only set the db if the result has been
 105     # wrapped (depending on query_args)
 106     if hasattr(result, 'set_db'):
 107         result.set_db(db)
 108
 109     return result