<s> import json <EOL> import time <EOL> import couchbase . _libcouchbase as _LCB <EOL> import couchbase . exceptions as exceptions <EOL> from couchbase . exceptions import CouchbaseError , ArgumentError <EOL> from couchbase . views . params import Query , SpatialQuery , STALE_OK <EOL> from couchbase . _pyport import single_dict_key <EOL> class BucketManager ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , cb ) : <EOL> self . _cb = cb <EOL> def _http_request ( self , ** kwargs ) : <EOL> return self . _cb . _http_request ( ** kwargs ) <EOL> def _mk_devmode ( self , * args ) : <EOL> return self . _cb . _mk_devmode ( * args ) <EOL> def _view ( self , * args , ** kwargs ) : <EOL> return self . _cb . _view ( * args , ** kwargs ) <EOL> def _doc_rev ( self , res ) : <EOL> """<STR_LIT>""" <EOL> jstr = res . headers [ '<STR_LIT>' ] <EOL> jobj = json . loads ( jstr ) <EOL> return jobj [ '<STR_LIT>' ] <EOL> def _poll_vq_single ( self , dname , use_devmode , ddresp ) : <EOL> """<STR_LIT>""" <EOL> vname = None <EOL> query = None <EOL> v_mr = ddresp . get ( '<STR_LIT>' , { } ) <EOL> v_spatial = ddresp . get ( '<STR_LIT>' , { } ) <EOL> if v_mr : <EOL> vname = single_dict_key ( v_mr ) <EOL> query = Query ( ) <EOL> elif v_spatial : <EOL> vname = single_dict_key ( v_spatial ) <EOL> query = SpatialQuery ( ) <EOL> if not vname : <EOL> return False <EOL> query . stale = STALE_OK <EOL> query . limit = <NUM_LIT:1> <EOL> for r in self . _cb . query ( dname , vname , use_devmode = use_devmode , <EOL> query = query ) : <EOL> pass <EOL> return True <EOL> def _design_poll ( self , name , mode , oldres , timeout = <NUM_LIT:5> , use_devmode = False ) : <EOL> """<STR_LIT>""" <EOL> if not timeout : <EOL> return True <EOL> if timeout < <NUM_LIT:0> : <EOL> raise ArgumentError . pyexc ( "<STR_LIT>" ) <EOL> t_end = time . time ( ) + timeout <EOL> old_rev = None <EOL> if oldres : <EOL> old_rev = self . _doc_rev ( oldres ) <EOL> while time . time ( ) < t_end : <EOL> try : <EOL> cur_resp = self . design_get ( name , use_devmode = use_devmode ) <EOL> if old_rev and self . _doc_rev ( cur_resp ) == old_rev : <EOL> continue <EOL> try : <EOL> if not self . _poll_vq_single ( <EOL> name , use_devmode , cur_resp . value ) : <EOL> continue <EOL> return True <EOL> except CouchbaseError : <EOL> continue <EOL> except CouchbaseError : <EOL> if mode == '<STR_LIT>' : <EOL> return True <EOL> raise exceptions . TimeoutError . pyexc ( <EOL> "<STR_LIT>" ) <EOL> def design_create ( self , name , ddoc , use_devmode = True , syncwait = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> name = self . _cb . _mk_devmode ( name , use_devmode ) <EOL> fqname = "<STR_LIT>" . format ( name ) <EOL> if not isinstance ( ddoc , dict ) : <EOL> ddoc = json . loads ( ddoc ) <EOL> ddoc = ddoc . copy ( ) <EOL> ddoc [ '<STR_LIT>' ] = fqname <EOL> ddoc = json . dumps ( ddoc ) <EOL> existing = None <EOL> if syncwait : <EOL> try : <EOL> existing = self . design_get ( name , use_devmode = False ) <EOL> except CouchbaseError : <EOL> pass <EOL> ret = self . _cb . _http_request ( <EOL> type = _LCB . LCB_HTTP_TYPE_VIEW , path = fqname , <EOL> method = _LCB . LCB_HTTP_METHOD_PUT , post_data = ddoc , <EOL> content_type = "<STR_LIT:application/json>" ) <EOL> self . _design_poll ( name , '<STR_LIT>' , existing , syncwait , <EOL> use_devmode = use_devmode ) <EOL> return ret <EOL> def design_get ( self , name , use_devmode = True ) : <EOL> """<STR_LIT>""" <EOL> name = self . _mk_devmode ( name , use_devmode ) <EOL> existing = self . _http_request ( type = _LCB . LCB_HTTP_TYPE_VIEW , <EOL> path = "<STR_LIT>" + name , <EOL> method = _LCB . LCB_HTTP_METHOD_GET , <EOL> content_type = "<STR_LIT:application/json>" ) <EOL> return existing <EOL> def design_publish ( self , name , syncwait = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> existing = self . design_get ( name , use_devmode = True ) <EOL> rv = self . design_create ( name , existing . value , use_devmode = False , <EOL> syncwait = syncwait ) <EOL> self . design_delete ( name , use_devmode = True , <EOL> syncwait = syncwait ) <EOL> self . _design_poll ( name , '<STR_LIT>' , None , <EOL> timeout = syncwait , use_devmode = False ) <EOL> return rv <EOL> def design_delete ( self , name , use_devmode = True , syncwait = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> name = self . _mk_devmode ( name , use_devmode ) <EOL> existing = None <EOL> if syncwait : <EOL> try : <EOL> existing = self . design_get ( name , use_devmode = False ) <EOL> except CouchbaseError : <EOL> pass <EOL> ret = self . _http_request ( type = _LCB . LCB_HTTP_TYPE_VIEW , <EOL> path = "<STR_LIT>" + name , <EOL> method = _LCB . LCB_HTTP_METHOD_DELETE ) <EOL> self . _design_poll ( name , '<STR_LIT>' , existing , syncwait ) <EOL> return ret </s>
