Support sub-document and N1QL

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
 

couchbase_ffi/_cinit.py

@@ -19,10 +19,14 @@
 #include <libcouchbase/api3.h>
 #include <libcouchbase/views.h>
 #include <libcouchbase/n1ql.h>
+#include <libcouchbase/subdoc.h>
 
 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,16 +39,31 @@
 #include <libcouchbase/api3.h>
 #include <libcouchbase/views.h>
 #include <libcouchbase/n1ql.h>
+#include <libcouchbase/subdoc.h>
 
 void _Cb_set_key(void *cmd, const void *key, size_t nkey) {
     LCB_CMD_SET_KEY((lcb_CMDBASE*)cmd, key, nkey);
 }
 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,

couchbase_ffi/_libcouchbase.py

@@ -70,7 +70,8 @@ def _stage2_bootstrap():
         ValueResult,
         OperationResult,
         HttpResult,
-        AsyncResult
+        AsyncResult,
+        _SDResult,
     )
     from couchbase_ffi.n1ql import _N1QLParams
 

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:

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)

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

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!')