diff --git a/couchbase_ffi/__init__.py b/couchbase_ffi/__init__.py index 057818c..7a960a2 100644 --- a/couchbase_ffi/__init__.py +++ b/couchbase_ffi/__init__.py @@ -11,7 +11,7 @@ def _mk_imp_override(srcname, replacement): :param replacement: The object which should act as the replacement """ class DummyImporter(object): - def find_module(self, fullname, path): + def find_module(self, fullname, path=None): if fullname == srcname: return self diff --git a/couchbase_ffi/_cinit.py b/couchbase_ffi/_cinit.py index 1b20ab1..987cfa6 100644 --- a/couchbase_ffi/_cinit.py +++ b/couchbase_ffi/_cinit.py @@ -19,10 +19,14 @@ #include #include #include +#include void _Cb_set_key(void*,const void*, size_t); void _Cb_set_val(void*,const void*, size_t); +void _Cb_sdspec_set_path(void*, const void*, size_t); +void _Cb_sdspec_set_value(void*, const void*, size_t); void _Cb_do_callback(lcb_socket_t s, short events, lcb_ioE_callback cb, void *arg); +void _Cb_n1ql_query_initcmd(lcb_CMDN1QL*, const char*, const int, lcb_N1QLCALLBACK); void memset(void*,int,int); """ @@ -35,6 +39,7 @@ #include #include #include +#include void _Cb_set_key(void *cmd, const void *key, size_t nkey) { LCB_CMD_SET_KEY((lcb_CMDBASE*)cmd, key, nkey); @@ -42,9 +47,23 @@ void _Cb_set_val(void *cmd, const void *val, size_t nval) { LCB_CMD_SET_VALUE((lcb_CMDSTORE*)cmd, val, nval); } +void _Cb_sdspec_set_path(void *sdspec, const void *path, size_t npath) { + LCB_SDSPEC_SET_PATH((lcb_SDSPEC*)sdspec, path, npath); +} +void _Cb_sdspec_set_value(void *sdspec, const void *value, size_t nvalue) { + LCB_SDSPEC_SET_VALUE((lcb_SDSPEC*)sdspec, value, nvalue); +} void _Cb_do_callback(lcb_socket_t s, short events, lcb_ioE_callback cb, void *arg) { cb(s, events, arg); } +void _Cb_n1ql_query_initcmd(lcb_CMDN1QL *nq, const char *params, const int nparams, + lcb_N1QLCALLBACK callback) { + nq->content_type = "application/json"; + nq->callback = callback; + nq->query = params; + nq->nquery = nparams; +} + LIBCOUCHBASE_API lcb_error_t lcb_n1p_synctok_for(lcb_N1QLPARAMS *params, lcb_t instance, diff --git a/couchbase_ffi/_libcouchbase.py b/couchbase_ffi/_libcouchbase.py index 946c39a..feedc72 100644 --- a/couchbase_ffi/_libcouchbase.py +++ b/couchbase_ffi/_libcouchbase.py @@ -70,7 +70,8 @@ def _stage2_bootstrap(): ValueResult, OperationResult, HttpResult, - AsyncResult + AsyncResult, + _SDResult, ) from couchbase_ffi.n1ql import _N1QLParams diff --git a/couchbase_ffi/_rtconfig.py b/couchbase_ffi/_rtconfig.py index 1e64714..fb0460e 100644 --- a/couchbase_ffi/_rtconfig.py +++ b/couchbase_ffi/_rtconfig.py @@ -44,6 +44,8 @@ def __init__(self): self.fmt_auto = None self.pypy_mres_factory = None self.view_path_helper = None + self.sd_result_type = None + self.sd_multival_type = None def configure(self, key, value): if not hasattr(self, key): @@ -79,12 +81,15 @@ def exc_args(self, msg='Bad arguments provided', obj=None): def exc_enc(self, msg='Bad key/value encoding', obj=None): self.exc_common(PYCBC_EXC_ENCODING, msg, 0, objextra=obj) - def exc_lcb(self, rc, msg='Operational error'): + def exc_lcb(self, rc, msg='Operational error', obj=None): try: cls = self.lcb_errno_map[rc] except KeyError: cls = self.default_exception.rc_to_exctype(rc) - raise cls({'rc': rc, 'message': msg}) + params = {'rc': rc, 'message': msg} + if obj is not None: + params['objextra'] = obj + raise cls(params) def exc_lock(self, msg=None): if msg is None: diff --git a/couchbase_ffi/bucket.py b/couchbase_ffi/bucket.py index e4ae2db..3067559 100644 --- a/couchbase_ffi/bucket.py +++ b/couchbase_ffi/bucket.py @@ -11,6 +11,7 @@ from couchbase_ffi.http import HttpRequest from couchbase_ffi.iops import IOPSWrapper from couchbase_ffi.lcbcntl import CNTL_VTYPE_MAP +from couchbase_ffi.n1ql import N1QLResult from couchbase_ffi.bufmanager import BufManager from couchbase_ffi._rtconfig import ( PyCBC, pycbc_exc_enc, pycbc_exc_args, pycbc_exc_lcb) @@ -202,6 +203,7 @@ def __init__(self, connection_string=None, connstr=None, 'observe': ffi.callback(CALLBACK_DECL, self._observe_callback), 'stats': ffi.callback(CALLBACK_DECL, self._stats_callback), 'http': ffi.callback(CALLBACK_DECL, self._http_callback), + 'subdoc': ffi.callback(CALLBACK_DECL, self._subdoc_callback), '_default': ffi.callback(CALLBACK_DECL, self._default_callback), '_bootstrap': ffi.callback('void(lcb_t,lcb_error_t)', self._bootstrap_callback), @@ -225,7 +227,9 @@ def __init__(self, connection_string=None, connstr=None, '_chained_endure': executors.DurabilityChainExecutor(self), 'observe': executors.ObserveExecutor(self), 'stats': executors.StatsExecutor(self), - '_rget': executors.GetReplicaExecutor(self) + '_rget': executors.GetReplicaExecutor(self), + 'lookup_in': executors.LookupInExecutor(self), + 'mutate_in': executors.MutateInExecutor(self) } self._install_cb(C.LCB_CALLBACK_DEFAULT, '_default') @@ -237,6 +241,8 @@ def __init__(self, connection_string=None, connstr=None, self._install_cb(C.LCB_CALLBACK_OBSERVE, 'observe') self._install_cb(C.LCB_CALLBACK_STATS, 'stats') self._install_cb(C.LCB_CALLBACK_HTTP, 'http') + self._install_cb(C.LCB_CALLBACK_SDLOOKUP, 'subdoc') + self._install_cb(C.LCB_CALLBACK_SDMUTATE, 'subdoc') C.lcb_set_bootstrap_callback(self._lcbh, self._bound_cb['_bootstrap']) # Set our properties @@ -475,9 +481,10 @@ def _execute_multi(self, name, kv, **kwargs): finally: self._do_unlock() - _VALUE_METHS = ['upsert', 'insert', 'replace', 'append', 'prepend'] + _VALUE_METHS = ['upsert', 'insert', 'replace', 'append', 'prepend', + 'mutate_in'] _KEY_METHS = ['get', 'lock', 'touch', 'remove', 'counter', - 'observe', 'endure', '_rget', '_unlock'] + 'observe', 'endure', '_rget', '_unlock', 'lookup_in'] for name in _VALUE_METHS + _KEY_METHS: n_single = name @@ -506,9 +513,18 @@ def unlock(self, key, cas, **kwargs): # noinspection PyUnresolvedReferences _rgetix_multi = _rget_multi - def _view_request(self, design, view, options, include_docs): + def _view_request(self, design, view, options, _flags): self._chk_no_pipeline('View requests not valid in pipeline mode') - res = ViewResult(design, view, options, include_docs) + res = ViewResult(design, view, options, _flags) + mres = self._make_mres() + mres[None] = res + res._schedule(self, mres) + return mres + + def _n1ql_query(self, params, prepare=0, cross_bucket=0): + self._chk_no_pipeline('N1QL queries cannot be executed ' + 'in pipeline context') + res = N1QLResult(params, prepare, cross_bucket) mres = self._make_mres() mres[None] = res res._schedule(self, mres) @@ -610,6 +626,7 @@ def _chk_op_done(self, mres): mres.invoke() def _chain_endure(self, optype, mres, result, dur): + mres._remaining -= 1 persist_to, replicate_to = dur proc = self._executors['_chained_endure'] try: @@ -627,7 +644,7 @@ def _default_callback(self, *args): _, mres = self._callback_common(*args) self._chk_op_done(mres) - def _callback_common(self, _, cbtype, resp): + def _callback_common(self, _, cbtype, resp, safe_rc=0): mres = ffi.from_handle(resp.cookie) buf = bytes(ffi.buffer(resp.key, resp.nkey)) try: @@ -637,7 +654,7 @@ def _callback_common(self, _, cbtype, resp): raise pycbc_exc_enc(buf) result.rc = resp.rc - if resp.rc: + if resp.rc and resp.rc != safe_rc: mres._add_bad_rc(resp.rc, result) else: result.cas = resp.cas @@ -677,6 +694,45 @@ def _get_callback(self, instance, cbtype, resp): self._chk_op_done(mres) + def _subdoc_callback(self, instance, cbtype, resp): + result, mres = self._callback_common(instance, cbtype, resp, + safe_rc=C.LCB_SUBDOC_MULTI_FAILURE) + resp = ffi.cast('lcb_RESPSUBDOC*', resp) + cur = ffi.new('lcb_SDENTRY*') + vii = ffi.new('size_t*') + oix = 0 + while C.lcb_sdresult_next(resp, cur, vii): + + if cbtype == C.LCB_CALLBACK_SDMUTATE: + cur_index = cur.index + else: + cur_index = oix + oix += 1 + + if cur.status == C.LCB_SUCCESS and cur.nvalue != 0: + buf = bytes(ffi.buffer(cur.value, cur.nvalue)) + try: + value = self._tc.decode_value(buf, FMT_JSON) + except: + try: + raise pycbc_exc_enc(obj=buf) + except PyCBC.default_exception: + mres._add_err(sys.exc_info()) + break + else: + value = None + + if cur.status != C.LCB_SUCCESS: + if cbtype == C.LCB_CALLBACK_SDMUTATE or cur.status != C.LCB_SUBDOC_PATH_ENOENT: + spec = result._specs[cur_index] + try: + raise pycbc_exc_lcb(cur.status, 'Subcommand failure', spec) + except PyCBC.default_exception: + mres._add_err(sys.exc_info()) + + result._results.append((cur.status, value)) + self._chk_op_done(mres) + def _remove_callback(self, instance, cbtype, resp): _, mres = self._callback_common(instance, cbtype, resp) self._chk_op_done(mres) diff --git a/couchbase_ffi/constants.py b/couchbase_ffi/constants.py index 2ebf0eb..eae3c94 100644 --- a/couchbase_ffi/constants.py +++ b/couchbase_ffi/constants.py @@ -31,6 +31,18 @@ LCB_DURABILITY_ETOOMANY = 33 LCB_DUPLICATE_COMMANDS = 34 LCB_HTTP_ERROR = 59 +LCB_SUBDOC_PATH_ENOENT = 63 +LCB_SUBDOC_PATH_MISMATCH = 64 +LCB_SUBDOC_PATH_EINVAL = 65 +LCB_SUBDOC_DOC_E2DEEP = 67 +LCB_SUBDOC_VALUE_E2DEEP = 74 +LCB_SUBDOC_VALUE_CANTINSERT = 68 +LCB_SUBDOC_DOC_NOTJSON = 69 +LCB_SUBDOC_NUM_ERANGE = 70 +LCB_SUBDOC_BAD_DELTA = 71 +LCB_SUBDOC_PATH_EEXISTS = 72 +LCB_SUBDOC_MULTI_FAILURE = 73 +LCB_EMPTY_PATH = 76 LCB_ADD = 1 LCB_REPLACE = 2 LCB_SET = 3 @@ -68,7 +80,7 @@ FMT_UTF8 = 67108868 FMT_PICKLE = 16777217 FMT_LEGACY_MASK = 7 -FMT_COMMON_MASK = -16777216 +FMT_COMMON_MASK = 4278190080 OBS_PERSISTED = 1 OBS_FOUND = 0 OBS_NOTFOUND = 128 @@ -102,5 +114,19 @@ LCB_ERRTYPE_INPUT = 1 LCB_CNTL_OP_TIMEOUT = 0 LCB_CNTL_VIEW_TIMEOUT = 1 +LCB_CNTL_SSL_MODE = 34 +LCB_SSL_ENABLED = 1 +LCB_CNTL_N1QL_TIMEOUT = 61 LCB_CMDVIEWQUERY_F_INCLUDE_DOCS = 65536 LCB_CMDVIEWQUERY_F_SPATIAL = 262144 +LCB_SDCMD_REPLACE = 3 +LCB_SDCMD_DICT_ADD = 4 +LCB_SDCMD_DICT_UPSERT = 5 +LCB_SDCMD_ARRAY_ADD_FIRST = 6 +LCB_SDCMD_ARRAY_ADD_LAST = 7 +LCB_SDCMD_ARRAY_ADD_UNIQUE = 8 +LCB_SDCMD_EXISTS = 2 +LCB_SDCMD_GET = 1 +LCB_SDCMD_COUNTER = 10 +LCB_SDCMD_REMOVE = 11 +LCB_SDCMD_ARRAY_INSERT = 9 diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index 74d4c3f..e7049c5 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -5,12 +5,13 @@ from couchbase.items import ItemCollection from couchbase_ffi.result import (OperationResult, ValueResult) -from couchbase_ffi.constants import FMT_UTF8 +from couchbase_ffi.constants import FMT_JSON, FMT_UTF8 from couchbase_ffi._cinit import get_handle -from couchbase_ffi._rtconfig import pycbc_exc_lcb, pycbc_exc_enc, pycbc_exc_args +from couchbase_ffi._rtconfig import PyCBC, pycbc_exc_lcb, pycbc_exc_enc, pycbc_exc_args from couchbase_ffi.bufmanager import BufManager ffi, C = get_handle() +bm = BufManager(ffi) class Options(dict): @@ -343,7 +344,15 @@ def execute(self, kv, **kwargs): if not len(kv): raise ArgumentError.pyexc(obj=kv, message="No items in container") - if isinstance(kv, dict): + if isinstance(kv, tuple) and len(kv) == 1 and isinstance(kv[0], dict): + # For sub-document specs + kv = kv[0] + is_dict = True + try: + kviter = kv.iteritems() + except AttributeError: + kviter = iter(kv.items()) + elif isinstance(kv, dict): is_dict = True try: kviter = kv.iteritems() @@ -544,6 +553,63 @@ def submit_single(self, c_key, c_len, value, item, key_options, global_options, return C.lcb_rget3(self.instance, mres._cdata, self.c_command) +class LookupInExecutor(BaseExecutor): + STRUCTNAME = 'lcb_CMDSUBDOC' + VALUES_ALLOWED = True + + def make_result(self, key, specs): + sr = PyCBC.sd_result_type() + sr.key = key + sr._specs = specs + return sr + + def convert_spec(self, spec, sdspec): + op = spec[0] + path = spec[1] + flags = spec[2] + path = self.parent._tc.encode_key(path) + sdspec.sdcmd = op + sdspec.options = flags + c_path, c_len = bm.new_cbuf(path) + C._Cb_sdspec_set_path(sdspec, c_path, c_len) + + def submit_single(self, c_key, c_len, specs, item, key_options, global_options, mres): + C._Cb_set_key(self.c_command, c_key, c_len) + nspecs = len(specs) + sdspecs = ffi.new('lcb_SDSPEC[]', nspecs) + self.c_command.specs = sdspecs + self.c_command.nspecs = nspecs + for x in range(nspecs): + spec, sdspec = specs[x], sdspecs[x] + self.convert_spec(spec, ffi.addressof(sdspec)) + return C.lcb_subdoc3(self.instance, mres._cdata, self.c_command) + + +class MutateInExecutor(LookupInExecutor): + + def convert_spec(self, spec, sdspec): + super(MutateInExecutor, self).convert_spec(spec, sdspec) + if len(spec) < 4: + return + value = spec[3] + value, __ = self.parent._tc.encode_value(value, FMT_JSON) + op = spec[0] + if op in [C.LCB_SDCMD_ARRAY_ADD_FIRST, C.LCB_SDCMD_ARRAY_ADD_LAST, + C.LCB_SDCMD_ARRAY_INSERT]: + # Strip outer [] for array operations + if not value.startswith('[') or not value.endswith(']'): + raise ValueFormatError('Serialized MultiValue shows ' + 'invalid JSON (maybe empty?)') + value = value[1:-1] + c_value, c_len = bm.new_cbuf(value) + C._Cb_sdspec_set_value(sdspec, c_value, c_len) + + def submit_single(self, c_key, c_len, specs, item, key_options, global_options, mres): + self.c_command.cas = get_cas(key_options, global_options, item) + self.c_command.exptime = get_ttl(key_options, global_options, item) + self.c_command.cmdflags |= global_options.get('_sd_doc_flags', 0) + base = super(MutateInExecutor, self) + return base.submit_single(c_key, c_len, specs, item, key_options, global_options, mres) class LockExecutor(GetExecutor): @@ -739,7 +805,6 @@ class StatsExecutor(BaseExecutor): STRUCTNAME = 'lcb_CMDSTATS' def __run_stat(self, k, mres): - bm = BufManager(ffi) if k: if not isinstance(k, basestring): raise pycbc_exc_args('Stats arguments must be strings only!') diff --git a/couchbase_ffi/http.py b/couchbase_ffi/http.py index 90e4a03..63327d0 100644 --- a/couchbase_ffi/http.py +++ b/couchbase_ffi/http.py @@ -88,3 +88,14 @@ def _handle_response(self, mres, resp): mres._add_err(sys.exc_info()) result.http_data = buf + if self._parent._is_async: + try: + mres._maybe_throw() + except: + mres.errback(mres, *sys.exc_info()) + else: + cb = mres.callback + if cb: + cb(mres) + finally: + del self._parent diff --git a/couchbase_ffi/n1ql.py b/couchbase_ffi/n1ql.py index 7525c41..db29a28 100644 --- a/couchbase_ffi/n1ql.py +++ b/couchbase_ffi/n1ql.py @@ -1,9 +1,11 @@ from couchbase_ffi._cinit import get_handle -from couchbase_ffi._rtconfig import pycbc_exc_lcb +from couchbase_ffi._rtconfig import pycbc_exc_lcb, PyCBC from couchbase_ffi.bufmanager import BufManager +from couchbase_ffi.view import buf2str, ViewResultBase ffi, C = get_handle() + class _N1QLParams(object): def __init__(self): self._lp = ffi.gc(C.lcb_n1p_new(), C.lcb_n1p_free) @@ -39,4 +41,33 @@ def add_posarg(self, arg): raise pycbc_exc_lcb(rc) def clear(self): - C.lcb_n1p_reset(self._lp) \ No newline at end of file + C.lcb_n1p_reset(self._lp) + + +class N1QLResult(ViewResultBase): + COMMAND_DECL = 'lcb_CMDN1QL*' + HANDLE_DECL = 'lcb_N1QLHANDLE*' + ROWCB_DECL = 'void(lcb_t,int,const lcb_RESPN1QL*)' + + def __init__(self, params, prepare=0, cross_bucket=0): + super(N1QLResult, self).__init__() + self._params = params + self._prepare = prepare + self._cross_bucket = cross_bucket + + def _init_command(self): + bm = BufManager(ffi) + cmd = self._c_command + C._Cb_n1ql_query_initcmd(cmd, bm.new_cstr(self._params), + len(self._params), self._bound_cb) + cmd.handle = self._c_handle + + def _query(self, parent, mres): + return C.lcb_n1ql_query(parent._lcbh, mres._cdata, self._c_command) + + def _handle_resp(self, resp, mres): + pass + + def _process_resp(self, resp, mres): + if resp.nrow: + return PyCBC.json_decode(buf2str(resp.row, resp.nrow)) \ No newline at end of file diff --git a/couchbase_ffi/result.py b/couchbase_ffi/result.py index adfb633..9eeda1f 100644 --- a/couchbase_ffi/result.py +++ b/couchbase_ffi/result.py @@ -49,6 +49,16 @@ def __init__(self): self.flags = 0 +class _SDResult(OperationResult): + # __slots__ = ['cas'] + _fldprops = PYCBC_RESFLD_KEY | PYCBC_RESFLD_CAS + + def __init__(self): + super(_SDResult, self).__init__() + self._results = [] + self._specs = [] + + class Item(ValueResult): def __getattr__(self, item): # This is needed because in C we just check the C field; however diff --git a/couchbase_ffi/view.py b/couchbase_ffi/view.py index 64da402..09a4f34 100644 --- a/couchbase_ffi/view.py +++ b/couchbase_ffi/view.py @@ -8,8 +8,6 @@ ffi, C = get_handle() -ROWCB_DECL = 'void(lcb_t,int,const lcb_RESPVIEWQUERY*)' - def mres2vres(mres): return mres[None] @@ -19,65 +17,37 @@ def buf2str(v, n): return from_cstring(ffi.cast('const char*', v), n) -class ViewResult(Result): - def __init__(self, ddoc, view, options, include_docs=False): - self._c_command = ffi.new('lcb_CMDVIEWQUERY*') - self._c_handle = ffi.new('lcb_VIEWHANDLE*') - self._ddoc = ddoc - self._view = view - self._options = options - self._include_docs = include_docs +class ViewResultBase(Result): + COMMAND_DECL = NotImplemented + HANDLE_DECL = NotImplemented + ROWCB_DECL = NotImplemented + + def __init__(self): + self._c_command = ffi.new(self.COMMAND_DECL) + self._c_handle = ffi.new(self.HANDLE_DECL) self._parent = None self.rows = [] self._rows_per_call = 0 - self._bound_cb = ffi.callback(ROWCB_DECL, self._on_single_row) + self._bound_cb = ffi.callback(self.ROWCB_DECL, self._on_single_row) self.done = False self.value = None self.http_status = 0 - @property - def key(self): - return 'VIEW[{0}/{1}]'.format(self._ddoc, self._view) - - @property - def rows_per_call(self): - return self._rows_per_call - @rows_per_call.setter - def rows_per_call(self, val): - self._rows_per_call = int(val) - def _schedule(self, parent, mres): - bm = BufManager(ffi) - urlopts = ffi.NULL - pypost = None - cmd = self._c_command - - if self._options: - in_uri, in_post = self._options._long_query_encoded - # Note, encoded means URI/JSON encoded; not charset - urlopts = bm.new_cstr(in_uri) - if in_post and in_post != '{}': - pypost = in_post - - C.lcb_view_query_initcmd( - cmd, bm.new_cstr(self._ddoc), bm.new_cstr(self._view), - urlopts, self._bound_cb) - - if pypost: - cmd.postdata, cmd.npostdata = bm.new_cbuf(pypost) - - if self._include_docs: - cmd.cmdflags |= C.LCB_CMDVIEWQUERY_F_INCLUDE_DOCS - - self._c_command.handle = self._c_handle - + self._init_command() self._parent = parent - rc = C.lcb_view_query(parent._lcbh, mres._cdata, self._c_command) + rc = self._query(parent, mres) if rc: raise pycbc_exc_lcb(rc) + def _init_command(self): + raise NotImplementedError + + def _query(self, parent, mres): + # should return rc + raise NotImplementedError + def _handle_done(self, resp, mres): - self.done = True self._c_handle = None if resp.rc: if resp.rc == C.LCB_HTTP_ERROR: @@ -88,18 +58,18 @@ def _handle_done(self, resp, mres): else: mres._add_bad_rc(resp.rc, self) - if resp.nvalue: - self.value = buf2str(resp.value, resp.nvalue) - try: - self.value = PyCBC.json_decode(self.value) - except: - pass + self._handle_resp(resp, mres) if resp.htresp: if not self.value and resp.htresp.nbody: self.value = buf2str(resp.htresp.body, resp.htresp.nbody) self.http_status = resp.htresp.htstatus + if self._parent._is_async: + self._invoke_async(mres, is_final=True) + + self.done = True + if self._parent._is_async: try: mres._maybe_throw() @@ -109,6 +79,16 @@ def _handle_done(self, resp, mres): finally: del self._parent + def _handle_resp(self, resp, mres): + raise NotImplementedError + + @property + def rows_per_call(self): + return self._rows_per_call + @rows_per_call.setter + def rows_per_call(self, val): + self._rows_per_call = int(val) + def _should_call(self, is_final): if is_final: return True @@ -139,6 +119,78 @@ def _on_single_row(self, instance, cbtype, resp): mres._add_bad_rc(resp.rc, self) return + row = self._process_resp(resp, mres) + + if row is not None: + # So now that we have a row.. + self.rows.append(row) + + if self._parent._is_async: + self._invoke_async(mres) + + def _process_resp(self, resp, mres): + raise NotImplementedError + + def fetch(self, mres): + C.lcb_wait(self._parent._lcbh) + ret = self.rows + self.rows = [] + mres._maybe_throw() + return ret + + +class ViewResult(ViewResultBase): + COMMAND_DECL = 'lcb_CMDVIEWQUERY*' + HANDLE_DECL = 'lcb_VIEWHANDLE*' + ROWCB_DECL = 'void(lcb_t,int,const lcb_RESPVIEWQUERY*)' + + def __init__(self, ddoc, view, options, flags): + super(ViewResult, self).__init__() + self._ddoc = ddoc + self._view = view + self._options = options + self._flags = flags + + @property + def key(self): + return 'VIEW[{0}/{1}]'.format(self._ddoc, self._view) + + def _init_command(self): + cmd = self._c_command + + bm = BufManager(ffi) + urlopts = ffi.NULL + pypost = None + + if self._options: + in_uri, in_post = self._options._long_query_encoded + # Note, encoded means URI/JSON encoded; not charset + urlopts = bm.new_cstr(in_uri) + if in_post and in_post != '{}': + pypost = in_post + + C.lcb_view_query_initcmd( + cmd, bm.new_cstr(self._ddoc), + bm.new_cstr(self._view), urlopts, self._bound_cb) + + if pypost: + cmd.postdata, cmd.npostdata = bm.new_cbuf(pypost) + + cmd.cmdflags = self._flags + cmd.handle = self._c_handle + + def _query(self, parent, mres): + return C.lcb_view_query(parent._lcbh, mres._cdata, self._c_command) + + def _handle_resp(self, resp, mres): + if resp.nvalue: + self.value = buf2str(resp.value, resp.nvalue) + try: + self.value = PyCBC.json_decode(self.value) + except: + pass + + def _process_resp(self, resp, mres): row = {} if resp.nkey: row['key'] = PyCBC.json_decode(buf2str(resp.key, resp.nkey)) @@ -147,7 +199,6 @@ def _on_single_row(self, instance, cbtype, resp): if resp.docid: # Document ID is always a simple string, so no need to decode row['id'] = buf2str(resp.docid, resp.ndocid) - if resp.docresp: py_doc = ValueResult() l_doc = resp.docresp @@ -163,17 +214,4 @@ def _on_single_row(self, instance, cbtype, resp): py_doc.value = tc.decode_value(buf, py_doc.flags) except: py_doc.value = buf[::] - - # So now that we have a row.. - self.rows.append(row) - if self._parent._is_async: - self._invoke_async(mres) - - - def fetch(self, mres): - C.lcb_wait(self._parent._lcbh) - ret = self.rows - self.rows = [] - mres._maybe_throw() - return ret - + return row