From f2868c2aebcb49245e1b8324a18881bb238aa631 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 19:29:19 +0900 Subject: [PATCH 01/26] Add missing error constants --- couchbase_ffi/constants.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/couchbase_ffi/constants.py b/couchbase_ffi/constants.py index 2ebf0eb..82663ee 100644 --- a/couchbase_ffi/constants.py +++ b/couchbase_ffi/constants.py @@ -31,6 +31,19 @@ LCB_DURABILITY_ETOOMANY = 33 LCB_DUPLICATE_COMMANDS = 34 LCB_HTTP_ERROR = 59 +LCB_SUBDOC_PATH_ENOENT = 0x3F +LCB_SUBDOC_PATH_MISMATCH = 0x40 +LCB_SUBDOC_PATH_EINVAL = 0x41 +LCB_SUBDOC_PATH_E2BIG = 0x42 +LCB_SUBDOC_DOC_E2DEEP = 0x43 +LCB_SUBDOC_VALUE_CANTINSERT = 0x44 +LCB_SUBDOC_DOC_NOTJSON = 0x45 +LCB_SUBDOC_NUM_ERANGE = 0x46 +LCB_SUBDOC_BAD_DELTA = 0x47 +LCB_SUBDOC_PATH_EEXISTS = 0x48 +LCB_SUBDOC_MULTI_FAILURE = 0x49 +LCB_SUBDOC_VALUE_E2DEEP = 0x4A +LCB_EMPTY_PATH = 0x4c LCB_ADD = 1 LCB_REPLACE = 2 LCB_SET = 3 From 5fb4113dea6bc422d91b846ee2ec91940a284f1d Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 19:43:23 +0900 Subject: [PATCH 02/26] Introduce _SDResult --- couchbase_ffi/_libcouchbase.py | 3 ++- couchbase_ffi/executors.py | 2 +- couchbase_ffi/result.py | 10 ++++++++++ 3 files changed, 13 insertions(+), 2 deletions(-) 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/executors.py b/couchbase_ffi/executors.py index 74d4c3f..d6712d9 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -4,7 +4,7 @@ from couchbase.exceptions import ValueFormatError, ArgumentError, CouchbaseError from couchbase.items import ItemCollection -from couchbase_ffi.result import (OperationResult, ValueResult) +from couchbase_ffi.result import (OperationResult, ValueResult, _SDResult) from couchbase_ffi.constants import FMT_UTF8 from couchbase_ffi._cinit import get_handle from couchbase_ffi._rtconfig import pycbc_exc_lcb, pycbc_exc_enc, pycbc_exc_args 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 From 7cd0d48c047ff4d1dde8f1be717589f34d724bb2 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 19:53:39 +0900 Subject: [PATCH 03/26] Generate constants.py by dump_constants() The Couchbase Python client version is 2.2.4. --- couchbase_ffi/constants.py | 41 +++++++++++++++++++++++++------------- 1 file changed, 27 insertions(+), 14 deletions(-) diff --git a/couchbase_ffi/constants.py b/couchbase_ffi/constants.py index 82663ee..eae3c94 100644 --- a/couchbase_ffi/constants.py +++ b/couchbase_ffi/constants.py @@ -31,19 +31,18 @@ LCB_DURABILITY_ETOOMANY = 33 LCB_DUPLICATE_COMMANDS = 34 LCB_HTTP_ERROR = 59 -LCB_SUBDOC_PATH_ENOENT = 0x3F -LCB_SUBDOC_PATH_MISMATCH = 0x40 -LCB_SUBDOC_PATH_EINVAL = 0x41 -LCB_SUBDOC_PATH_E2BIG = 0x42 -LCB_SUBDOC_DOC_E2DEEP = 0x43 -LCB_SUBDOC_VALUE_CANTINSERT = 0x44 -LCB_SUBDOC_DOC_NOTJSON = 0x45 -LCB_SUBDOC_NUM_ERANGE = 0x46 -LCB_SUBDOC_BAD_DELTA = 0x47 -LCB_SUBDOC_PATH_EEXISTS = 0x48 -LCB_SUBDOC_MULTI_FAILURE = 0x49 -LCB_SUBDOC_VALUE_E2DEEP = 0x4A -LCB_EMPTY_PATH = 0x4c +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 @@ -81,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 @@ -115,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 From a6c9603d3ebccfa1fcc14ac1ca36cbde546cbef1 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 19:54:29 +0900 Subject: [PATCH 04/26] Define sd types in _PyCBC_Class --- couchbase_ffi/_rtconfig.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/couchbase_ffi/_rtconfig.py b/couchbase_ffi/_rtconfig.py index 1e64714..2eb8606 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): From bef3bc8242a89c5c70bf37e5dc806578b11eac43 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 20:52:25 +0900 Subject: [PATCH 05/26] [WIP] Start to make LookupInExecutor --- couchbase_ffi/bucket.py | 5 +++-- couchbase_ffi/executors.py | 12 ++++++++++++ 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/couchbase_ffi/bucket.py b/couchbase_ffi/bucket.py index e4ae2db..82a729d 100644 --- a/couchbase_ffi/bucket.py +++ b/couchbase_ffi/bucket.py @@ -225,7 +225,8 @@ 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) } self._install_cb(C.LCB_CALLBACK_DEFAULT, '_default') @@ -477,7 +478,7 @@ def _execute_multi(self, name, kv, **kwargs): _VALUE_METHS = ['upsert', 'insert', 'replace', 'append', 'prepend'] _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 diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index d6712d9..1eb28b0 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -343,6 +343,9 @@ def execute(self, kv, **kwargs): if not len(kv): raise ArgumentError.pyexc(obj=kv, message="No items in container") + if isinstance(kv, tuple): + kv = kv[0] + if isinstance(kv, dict): is_dict = True try: @@ -544,6 +547,15 @@ 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, value): + sr = _SDResult() + sr.key = key + sr._specs = value + return sr class LockExecutor(GetExecutor): From 1e16cf6d3bd46dfb4ea2eb2effb6d73ff98fa2a6 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 22:35:35 +0900 Subject: [PATCH 06/26] Implement lookup_in for single spec --- couchbase_ffi/_cinit.py | 6 ++++++ couchbase_ffi/bucket.py | 13 +++++++++++++ couchbase_ffi/executors.py | 28 +++++++++++++++++++++++----- 3 files changed, 42 insertions(+), 5 deletions(-) diff --git a/couchbase_ffi/_cinit.py b/couchbase_ffi/_cinit.py index 1b20ab1..1a4b8cd 100644 --- a/couchbase_ffi/_cinit.py +++ b/couchbase_ffi/_cinit.py @@ -19,9 +19,11 @@ #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_do_callback(lcb_socket_t s, short events, lcb_ioE_callback cb, void *arg); void memset(void*,int,int); """ @@ -35,6 +37,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,6 +45,9 @@ 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_do_callback(lcb_socket_t s, short events, lcb_ioE_callback cb, void *arg) { cb(s, events, arg); } diff --git a/couchbase_ffi/bucket.py b/couchbase_ffi/bucket.py index 82a729d..795551d 100644 --- a/couchbase_ffi/bucket.py +++ b/couchbase_ffi/bucket.py @@ -202,6 +202,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), + 'sdlookup': ffi.callback(CALLBACK_DECL, self._sdlookup_callback), '_default': ffi.callback(CALLBACK_DECL, self._default_callback), '_bootstrap': ffi.callback('void(lcb_t,lcb_error_t)', self._bootstrap_callback), @@ -238,6 +239,7 @@ 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, 'sdlookup') C.lcb_set_bootstrap_callback(self._lcbh, self._bound_cb['_bootstrap']) # Set our properties @@ -678,6 +680,17 @@ def _get_callback(self, instance, cbtype, resp): self._chk_op_done(mres) + def _sdlookup_callback(self, instance, cbtype, resp): + result, mres = self._callback_common(instance, cbtype, resp) + resp = ffi.cast('lcb_RESPSUBDOC*', resp) + cur = ffi.new('lcb_SDENTRY*') + vii = ffi.new('size_t*') + while C.lcb_sdresult_next(resp, cur, vii): + buf = bytes(ffi.buffer(cur.value, cur.nvalue)) + value = self._tc.decode_value(buf, FMT_JSON) + 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/executors.py b/couchbase_ffi/executors.py index 1eb28b0..576df53 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -4,10 +4,10 @@ from couchbase.exceptions import ValueFormatError, ArgumentError, CouchbaseError from couchbase.items import ItemCollection -from couchbase_ffi.result import (OperationResult, ValueResult, _SDResult) +from couchbase_ffi.result import (OperationResult, ValueResult) from couchbase_ffi.constants import 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() @@ -551,12 +551,30 @@ class LookupInExecutor(BaseExecutor): STRUCTNAME = 'lcb_CMDSUBDOC' VALUES_ALLOWED = True - def make_result(self, key, value): - sr = _SDResult() + def make_result(self, key, specs): + sr = PyCBC.sd_result_type() sr.key = key - sr._specs = value + sr._specs = specs return sr + def convert_spec(self, spec, sdspec): + op, path, flags = spec + path = self.parent._tc.encode_key(path) + sdspec.sdcmd = op # int + sdspec.options = flags # unsigned + bm = BufManager(ffi) + 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) + sdspec = ffi.new('lcb_SDSPEC*') + spec = specs[0] + self.c_command.specs = sdspec + self.c_command.nspecs = 1 + self.convert_spec(spec, sdspec) + return C.lcb_subdoc3(self.instance, mres._cdata, self.c_command) + class LockExecutor(GetExecutor): IS_LOCK = True From ec50d3535159e93eea89436158a7c67a8534c4c2 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 22:49:02 +0900 Subject: [PATCH 07/26] Implement multiple specs for lookup_in --- couchbase_ffi/executors.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index 576df53..95b4f76 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -568,11 +568,13 @@ def convert_spec(self, spec, sdspec): 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) - sdspec = ffi.new('lcb_SDSPEC*') - spec = specs[0] - self.c_command.specs = sdspec - self.c_command.nspecs = 1 - self.convert_spec(spec, sdspec) + 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) From a7bde7d402954c9a32b3bdced8c32926bb8eda52 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 23:16:31 +0900 Subject: [PATCH 08/26] Implement mutate_in Don't implement mutate_in with mutiple values yet. --- couchbase_ffi/_cinit.py | 4 ++++ couchbase_ffi/bucket.py | 20 +++++++++++++------- couchbase_ffi/executors.py | 29 +++++++++++++++++++++++++---- 3 files changed, 42 insertions(+), 11 deletions(-) diff --git a/couchbase_ffi/_cinit.py b/couchbase_ffi/_cinit.py index 1a4b8cd..85d6ac0 100644 --- a/couchbase_ffi/_cinit.py +++ b/couchbase_ffi/_cinit.py @@ -24,6 +24,7 @@ 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 memset(void*,int,int); """ @@ -48,6 +49,9 @@ 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); } diff --git a/couchbase_ffi/bucket.py b/couchbase_ffi/bucket.py index 795551d..56ad951 100644 --- a/couchbase_ffi/bucket.py +++ b/couchbase_ffi/bucket.py @@ -202,7 +202,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), - 'sdlookup': ffi.callback(CALLBACK_DECL, self._sdlookup_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), @@ -227,7 +227,8 @@ def __init__(self, connection_string=None, connstr=None, 'observe': executors.ObserveExecutor(self), 'stats': executors.StatsExecutor(self), '_rget': executors.GetReplicaExecutor(self), - 'lookup_in': executors.LookupInExecutor(self) + 'lookup_in': executors.LookupInExecutor(self), + 'mutate_in': executors.MutateInExecutor(self) } self._install_cb(C.LCB_CALLBACK_DEFAULT, '_default') @@ -239,7 +240,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, 'sdlookup') + 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 @@ -478,7 +480,8 @@ 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', 'lookup_in'] @@ -680,14 +683,17 @@ def _get_callback(self, instance, cbtype, resp): self._chk_op_done(mres) - def _sdlookup_callback(self, instance, cbtype, resp): + def _subdoc_callback(self, instance, cbtype, resp): result, mres = self._callback_common(instance, cbtype, resp) resp = ffi.cast('lcb_RESPSUBDOC*', resp) cur = ffi.new('lcb_SDENTRY*') vii = ffi.new('size_t*') while C.lcb_sdresult_next(resp, cur, vii): - buf = bytes(ffi.buffer(cur.value, cur.nvalue)) - value = self._tc.decode_value(buf, FMT_JSON) + if cur.status == C.LCB_SUCCESS and cur.nvalue != 0: + buf = bytes(ffi.buffer(cur.value, cur.nvalue)) + value = self._tc.decode_value(buf, FMT_JSON) + else: + value = None result._results.append((cur.status, value)) self._chk_op_done(mres) diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index 95b4f76..01b7863 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -5,7 +5,7 @@ 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, pycbc_exc_lcb, pycbc_exc_enc, pycbc_exc_args from couchbase_ffi.bufmanager import BufManager @@ -558,10 +558,12 @@ def make_result(self, key, specs): return sr def convert_spec(self, spec, sdspec): - op, path, flags = spec + op = spec[0] + path = spec[1] + flags = spec[2] path = self.parent._tc.encode_key(path) - sdspec.sdcmd = op # int - sdspec.options = flags # unsigned + sdspec.sdcmd = op + sdspec.options = flags bm = BufManager(ffi) c_path, c_len = bm.new_cbuf(path) C._Cb_sdspec_set_path(sdspec, c_path, c_len) @@ -578,6 +580,25 @@ def submit_single(self, c_key, c_len, specs, item, key_options, global_options, 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) + value = spec[3] + value, __ = self.parent._tc.encode_value(value, FMT_JSON) + bm = BufManager(ffi) + c_value, c_len = bm.new_cbuf(value) + op = spec[0] + if op in [C.LCB_SDCMD_ARRAY_ADD_FIRST, + C.LCB_SDCMD_ARRAY_ADD_LAST, + C.LCB_SDCMD_ARRAY_INSERT]: + # use multival + # NOTE: not implemented yet + pass + + C._Cb_sdspec_set_value(sdspec, c_value, c_len) + + class LockExecutor(GetExecutor): IS_LOCK = True From 671867ec477a4ad07a165b759cda547f3953bf80 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 23:22:32 +0900 Subject: [PATCH 09/26] Share BufManager --- couchbase_ffi/executors.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index 01b7863..c1254e2 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -11,6 +11,7 @@ from couchbase_ffi.bufmanager import BufManager ffi, C = get_handle() +bm = BufManager(ffi) class Options(dict): @@ -564,7 +565,6 @@ def convert_spec(self, spec, sdspec): path = self.parent._tc.encode_key(path) sdspec.sdcmd = op sdspec.options = flags - bm = BufManager(ffi) c_path, c_len = bm.new_cbuf(path) C._Cb_sdspec_set_path(sdspec, c_path, c_len) @@ -586,7 +586,6 @@ def convert_spec(self, spec, sdspec): super(MutateInExecutor, self).convert_spec(spec, sdspec) value = spec[3] value, __ = self.parent._tc.encode_value(value, FMT_JSON) - bm = BufManager(ffi) c_value, c_len = bm.new_cbuf(value) op = spec[0] if op in [C.LCB_SDCMD_ARRAY_ADD_FIRST, @@ -792,7 +791,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!') From 3bc2e47ba94a7e5a620cdadfaa4c5506339421d8 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 23:24:55 +0900 Subject: [PATCH 10/26] Strip outer [] for array sub-document operations --- couchbase_ffi/executors.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index c1254e2..204e52a 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -586,15 +586,15 @@ def convert_spec(self, spec, sdspec): super(MutateInExecutor, self).convert_spec(spec, sdspec) value = spec[3] value, __ = self.parent._tc.encode_value(value, FMT_JSON) - c_value, c_len = bm.new_cbuf(value) op = spec[0] - if op in [C.LCB_SDCMD_ARRAY_ADD_FIRST, - C.LCB_SDCMD_ARRAY_ADD_LAST, + if op in [C.LCB_SDCMD_ARRAY_ADD_FIRST, C.LCB_SDCMD_ARRAY_ADD_LAST, C.LCB_SDCMD_ARRAY_INSERT]: - # use multival - # NOTE: not implemented yet - pass - + # 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) From 8a48ff16390f812d5859d5835c7b217d3fb98091 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 23:33:01 +0900 Subject: [PATCH 11/26] Just trigger Travis CI From 476fc0fa8a2806bac5c28033594863731fba7a3a Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 23:46:44 +0900 Subject: [PATCH 12/26] path parameter for find_module() is optional --- couchbase_ffi/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 From 3edfd8b31a41903a997bec1a5302d502f6afa8f5 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 23:54:42 +0900 Subject: [PATCH 13/26] Don't slice key for GET --- couchbase_ffi/executors.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index 204e52a..13fefb9 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -344,10 +344,15 @@ def execute(self, kv, **kwargs): if not len(kv): raise ArgumentError.pyexc(obj=kv, message="No items in container") - if isinstance(kv, tuple): + if isinstance(kv, tuple) and len(kv) == 1 and isinstance(kv[0], dict): + # For sub-document specs kv = kv[0] - - if isinstance(kv, dict): + is_dict = True + try: + kviter = kv.iteritems() + except AttributeError: + kviter = iter(kv.items()) + elif isinstance(kv, dict): is_dict = True try: kviter = kv.iteritems() From 04fb9138fe8abd4e24a832bf93a595963caec126 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Thu, 1 Jun 2017 23:55:33 +0900 Subject: [PATCH 14/26] Capitalize code comment It looks like a convention. --- couchbase_ffi/executors.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index 13fefb9..d925983 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -594,7 +594,7 @@ def convert_spec(self, spec, sdspec): 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 + # Strip outer [] for array operations if not value.startswith('[') or not value.endswith(']'): raise ValueFormatError('Serialized MultiValue shows ' 'invalid JSON (maybe empty?)') From 5cd81874e6f28b921751c2f0aaa07875c5d0eaa5 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Fri, 2 Jun 2017 00:13:21 +0900 Subject: [PATCH 15/26] _view_request() takes _flags instead of include_docs --- couchbase_ffi/bucket.py | 4 ++-- couchbase_ffi/view.py | 7 +++---- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/couchbase_ffi/bucket.py b/couchbase_ffi/bucket.py index 56ad951..a14d1f5 100644 --- a/couchbase_ffi/bucket.py +++ b/couchbase_ffi/bucket.py @@ -512,9 +512,9 @@ 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) diff --git a/couchbase_ffi/view.py b/couchbase_ffi/view.py index 64da402..cb36fd0 100644 --- a/couchbase_ffi/view.py +++ b/couchbase_ffi/view.py @@ -20,13 +20,13 @@ def buf2str(v, n): class ViewResult(Result): - def __init__(self, ddoc, view, options, include_docs=False): + def __init__(self, ddoc, view, options, flags): 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 + self._flags = flags self._parent = None self.rows = [] self._rows_per_call = 0 @@ -66,8 +66,7 @@ def _schedule(self, parent, mres): if pypost: cmd.postdata, cmd.npostdata = bm.new_cbuf(pypost) - if self._include_docs: - cmd.cmdflags |= C.LCB_CMDVIEWQUERY_F_INCLUDE_DOCS + cmd.cmdflags = self._flags self._c_command.handle = self._c_handle From 149950f9e2de319225d72b7874acfe042ed8e56b Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Fri, 2 Jun 2017 01:07:31 +0900 Subject: [PATCH 16/26] Invoke View final callback before done mark Because GView fails with this error: ``` Traceback (most recent call last): ... File "gcouchbase/bucket.py", line 60, in __iter__ for row in self._process_payload(rowset): File "couchbase/views/iterator.py", line 366, in _process_payload return self.row_processor.handle_rows(rows, self._parent, False) AttributeError: 'GView' object has no attribute '_parent' ``` --- couchbase_ffi/view.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/couchbase_ffi/view.py b/couchbase_ffi/view.py index cb36fd0..61d0b6a 100644 --- a/couchbase_ffi/view.py +++ b/couchbase_ffi/view.py @@ -76,7 +76,6 @@ def _schedule(self, parent, mres): raise pycbc_exc_lcb(rc) def _handle_done(self, resp, mres): - self.done = True self._c_handle = None if resp.rc: if resp.rc == C.LCB_HTTP_ERROR: @@ -99,6 +98,11 @@ def _handle_done(self, resp, mres): 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() From f52cecf1f63b4027e457a88a3dc5c2af94245690 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Mon, 5 Jun 2017 15:03:26 +0900 Subject: [PATCH 17/26] HttpRequest calls callback when async mode --- couchbase_ffi/http.py | 11 +++++++++++ 1 file changed, 11 insertions(+) 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 From 8ecb0045e25f5feaa7866658fd2051add4870d7b Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Mon, 5 Jun 2017 16:06:33 +0900 Subject: [PATCH 18/26] Implement N1QL It's very dirty but works. --- couchbase_ffi/_cinit.py | 9 +++ couchbase_ffi/bucket.py | 10 ++++ couchbase_ffi/n1ql.py | 130 +++++++++++++++++++++++++++++++++++++++- 3 files changed, 147 insertions(+), 2 deletions(-) diff --git a/couchbase_ffi/_cinit.py b/couchbase_ffi/_cinit.py index 85d6ac0..987cfa6 100644 --- a/couchbase_ffi/_cinit.py +++ b/couchbase_ffi/_cinit.py @@ -26,6 +26,7 @@ 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); """ @@ -55,6 +56,14 @@ 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/bucket.py b/couchbase_ffi/bucket.py index a14d1f5..43599dc 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) @@ -520,6 +521,15 @@ def _view_request(self, design, view, options, _flags): 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) + return mres + def _http_request(self, path, **kwargs): self._chk_no_pipeline('HTTP requests not valid in pipeline mode') htreq = HttpRequest(path, **kwargs) diff --git a/couchbase_ffi/n1ql.py b/couchbase_ffi/n1ql.py index 7525c41..ea62e04 100644 --- a/couchbase_ffi/n1ql.py +++ b/couchbase_ffi/n1ql.py @@ -1,9 +1,13 @@ 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._strutil import from_cstring from couchbase_ffi.bufmanager import BufManager +from couchbase_ffi.result import Result + ffi, C = get_handle() + class _N1QLParams(object): def __init__(self): self._lp = ffi.gc(C.lcb_n1p_new(), C.lcb_n1p_free) @@ -39,4 +43,126 @@ 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) + + +ROWCB_DECL = 'void(lcb_t,int,const lcb_RESPN1QL*)' + + +def buf2str(v, n): + return from_cstring(ffi.cast('const char*', v), n) + + +class N1QLResult(Result): + def __init__(self, params, prepare=0, cross_bucket=0): + self._c_command = ffi.new('lcb_CMDN1QL*') + self._c_handle = ffi.new('lcb_N1QLHANDLE*') + self._params = params + self._prepare = prepare + self._cross_bucket = cross_bucket + self._parent = None + self.rows = [] + + # self._rows_per_call = 0 + self._bound_cb = ffi.callback(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 + + C._Cb_n1ql_query_initcmd(cmd, bm.new_cstr(self._params), + len(self._params), self._bound_cb) + + # if pypost: + # cmd.postdata, cmd.npostdata = bm.new_cbuf(pypost) + + cmd.handle = self._c_handle + + self._parent = parent + rc = C.lcb_n1ql_query(parent._lcbh, mres._cdata, self._c_command) + if rc: + raise pycbc_exc_lcb(rc) + + def _handle_done(self, resp, mres): + self._c_handle = None + if resp.rc: + if resp.rc == C.LCB_HTTP_ERROR: + try: + raise PyCBC.exc_http(self.value) + except: + mres._add_err(sys.exc_info()) + else: + mres._add_bad_rc(resp.rc, self) + + 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() + self._invoke_async(mres, is_final=True) + except: + mres.errback(mres, *sys.exc_info()) + finally: + del self._parent + + def _should_call(self, is_final): + if is_final: + return True + return -1 < self._rows_per_call < len(self.rows) + + def _invoke_async(self, mres, is_final=False): + if not self._should_call(is_final=is_final): + return + + cb = mres.callback + if cb: + cb(mres) + self.rows = [] + + def _on_single_row(self, instance, cbtype, resp): + mres = ffi.from_handle(resp.cookie) + if resp.rflags & C.LCB_RESP_F_FINAL: + self._handle_done(resp, mres) + return + + if resp.rc != C.LCB_SUCCESS: + mres._add_bad_rc(resp.rc, self) + return + + if resp.nrow: + row = PyCBC.json_decode(buf2str(resp.row, resp.nrow)) + # 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 \ No newline at end of file From cb6bbcfe13c4d7f366a915bbd0935aec64b62371 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Mon, 5 Jun 2017 16:34:27 +0900 Subject: [PATCH 19/26] Introduce ViewResultBase It shares similar code between ViewResult and N1QLResult. --- couchbase_ffi/n1ql.py | 121 ++++-------------------------- couchbase_ffi/view.py | 169 +++++++++++++++++++++++++----------------- 2 files changed, 115 insertions(+), 175 deletions(-) diff --git a/couchbase_ffi/n1ql.py b/couchbase_ffi/n1ql.py index ea62e04..db29a28 100644 --- a/couchbase_ffi/n1ql.py +++ b/couchbase_ffi/n1ql.py @@ -1,9 +1,7 @@ from couchbase_ffi._cinit import get_handle from couchbase_ffi._rtconfig import pycbc_exc_lcb, PyCBC -from couchbase_ffi._strutil import from_cstring from couchbase_ffi.bufmanager import BufManager -from couchbase_ffi.result import Result - +from couchbase_ffi.view import buf2str, ViewResultBase ffi, C = get_handle() @@ -46,123 +44,30 @@ def clear(self): C.lcb_n1p_reset(self._lp) -ROWCB_DECL = 'void(lcb_t,int,const lcb_RESPN1QL*)' - - -def buf2str(v, n): - return from_cstring(ffi.cast('const char*', v), n) +class N1QLResult(ViewResultBase): + COMMAND_DECL = 'lcb_CMDN1QL*' + HANDLE_DECL = 'lcb_N1QLHANDLE*' + ROWCB_DECL = 'void(lcb_t,int,const lcb_RESPN1QL*)' - -class N1QLResult(Result): def __init__(self, params, prepare=0, cross_bucket=0): - self._c_command = ffi.new('lcb_CMDN1QL*') - self._c_handle = ffi.new('lcb_N1QLHANDLE*') + super(N1QLResult, self).__init__() self._params = params self._prepare = prepare self._cross_bucket = cross_bucket - self._parent = None - self.rows = [] - - # self._rows_per_call = 0 - self._bound_cb = ffi.callback(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): + def _init_command(self): bm = BufManager(ffi) - urlopts = ffi.NULL - pypost = None cmd = self._c_command - C._Cb_n1ql_query_initcmd(cmd, bm.new_cstr(self._params), len(self._params), self._bound_cb) - - # if pypost: - # cmd.postdata, cmd.npostdata = bm.new_cbuf(pypost) - cmd.handle = self._c_handle - self._parent = parent - rc = C.lcb_n1ql_query(parent._lcbh, mres._cdata, self._c_command) - if rc: - raise pycbc_exc_lcb(rc) - - def _handle_done(self, resp, mres): - self._c_handle = None - if resp.rc: - if resp.rc == C.LCB_HTTP_ERROR: - try: - raise PyCBC.exc_http(self.value) - except: - mres._add_err(sys.exc_info()) - else: - mres._add_bad_rc(resp.rc, self) - - 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() - self._invoke_async(mres, is_final=True) - except: - mres.errback(mres, *sys.exc_info()) - finally: - del self._parent + def _query(self, parent, mres): + return C.lcb_n1ql_query(parent._lcbh, mres._cdata, self._c_command) - def _should_call(self, is_final): - if is_final: - return True - return -1 < self._rows_per_call < len(self.rows) - - def _invoke_async(self, mres, is_final=False): - if not self._should_call(is_final=is_final): - return - - cb = mres.callback - if cb: - cb(mres) - self.rows = [] - - def _on_single_row(self, instance, cbtype, resp): - mres = ffi.from_handle(resp.cookie) - if resp.rflags & C.LCB_RESP_F_FINAL: - self._handle_done(resp, mres) - return - - if resp.rc != C.LCB_SUCCESS: - mres._add_bad_rc(resp.rc, self) - return + def _handle_resp(self, resp, mres): + pass + def _process_resp(self, resp, mres): if resp.nrow: - row = PyCBC.json_decode(buf2str(resp.row, resp.nrow)) - # 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 \ No newline at end of file + return PyCBC.json_decode(buf2str(resp.row, resp.nrow)) \ No newline at end of file diff --git a/couchbase_ffi/view.py b/couchbase_ffi/view.py index 61d0b6a..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,62 +17,36 @@ def buf2str(v, n): return from_cstring(ffi.cast('const char*', v), n) -class ViewResult(Result): - def __init__(self, ddoc, view, options, flags): - self._c_command = ffi.new('lcb_CMDVIEWQUERY*') - self._c_handle = ffi.new('lcb_VIEWHANDLE*') - self._ddoc = ddoc - self._view = view - self._options = options - self._flags = flags +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) - - cmd.cmdflags = self._flags - - 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._c_handle = None if resp.rc: @@ -86,12 +58,7 @@ 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: @@ -112,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 @@ -142,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)) @@ -150,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 @@ -166,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 From 05e4a27bc9c08d64b25ac2f0a0f452c0e365ad16 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Mon, 5 Jun 2017 17:43:14 +0900 Subject: [PATCH 20/26] Fix async hang on endure chain --- couchbase_ffi/bucket.py | 1 + 1 file changed, 1 insertion(+) diff --git a/couchbase_ffi/bucket.py b/couchbase_ffi/bucket.py index 43599dc..1880240 100644 --- a/couchbase_ffi/bucket.py +++ b/couchbase_ffi/bucket.py @@ -626,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: From 13471b88e787350c18dd02a8ea20be9f3fde3e96 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Tue, 6 Jun 2017 03:59:46 +0900 Subject: [PATCH 21/26] SubdocResult keeps LCB_SUBDOC_MULTI_FAILURE as harmless --- couchbase_ffi/bucket.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/couchbase_ffi/bucket.py b/couchbase_ffi/bucket.py index 1880240..598af54 100644 --- a/couchbase_ffi/bucket.py +++ b/couchbase_ffi/bucket.py @@ -644,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: @@ -654,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 @@ -695,7 +695,8 @@ 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) + 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*') From e339c43213f57a7d9404822cf9980547a67ee5b5 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Tue, 6 Jun 2017 17:46:15 +0900 Subject: [PATCH 22/26] Set command options for mutate_in But upsert_doc option is refused by the server. This behavior is same in CPython. --- couchbase_ffi/executors.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index d925983..4c5a831 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -602,6 +602,13 @@ def convert_spec(self, spec, sdspec): 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) + return super(MutateInExecutor, self).submit_single( + c_key, c_len, specs, item, key_options, global_options, mres) + class LockExecutor(GetExecutor): IS_LOCK = True From 4c7788c23659589d13810f9ff3adc516fd3a211d Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Tue, 6 Jun 2017 18:31:08 +0900 Subject: [PATCH 23/26] Copy Sub-document exception handling To pass test case: tests/test_compat.py::SubdocTest_FFI::test_counter_in --- couchbase_ffi/_rtconfig.py | 7 +++++-- couchbase_ffi/bucket.py | 26 +++++++++++++++++++++++++- 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/couchbase_ffi/_rtconfig.py b/couchbase_ffi/_rtconfig.py index 2eb8606..fb0460e 100644 --- a/couchbase_ffi/_rtconfig.py +++ b/couchbase_ffi/_rtconfig.py @@ -81,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 598af54..0b3c5ca 100644 --- a/couchbase_ffi/bucket.py +++ b/couchbase_ffi/bucket.py @@ -700,12 +700,36 @@ def _subdoc_callback(self, instance, cbtype, resp): 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: + oix += 1 + cur_index = oix + if cur.status == C.LCB_SUCCESS and cur.nvalue != 0: buf = bytes(ffi.buffer(cur.value, cur.nvalue)) - value = self._tc.decode_value(buf, FMT_JSON) + 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) From db21dc7186b13924e4d2d76fae58caaf8f914105 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Tue, 6 Jun 2017 18:32:23 +0900 Subject: [PATCH 24/26] Style --- couchbase_ffi/executors.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index 4c5a831..bdd43b5 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -606,8 +606,8 @@ def submit_single(self, c_key, c_len, specs, item, key_options, global_options, 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) - return super(MutateInExecutor, self).submit_single( - c_key, c_len, specs, item, key_options, global_options, mres) + base = super(MutateInExecutor, self) + return base.submit_single(c_key, c_len, specs, item, key_options, global_options, mres) class LockExecutor(GetExecutor): From dd9d6b700b9e2fe585c7d0ee3d80d14241a8331a Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Tue, 6 Jun 2017 18:34:28 +0900 Subject: [PATCH 25/26] Fix lookup_in index bug --- couchbase_ffi/bucket.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/couchbase_ffi/bucket.py b/couchbase_ffi/bucket.py index 0b3c5ca..3067559 100644 --- a/couchbase_ffi/bucket.py +++ b/couchbase_ffi/bucket.py @@ -706,8 +706,8 @@ def _subdoc_callback(self, instance, cbtype, resp): if cbtype == C.LCB_CALLBACK_SDMUTATE: cur_index = cur.index else: - oix += 1 cur_index = oix + oix += 1 if cur.status == C.LCB_SUCCESS and cur.nvalue != 0: buf = bytes(ffi.buffer(cur.value, cur.nvalue)) From b0b8864476d9cfd5df0c276d19996a6d2d8edcc6 Mon Sep 17 00:00:00 2001 From: Heungsub Lee Date: Tue, 6 Jun 2017 18:50:58 +0900 Subject: [PATCH 26/26] Fix set_remove failure --- couchbase_ffi/executors.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/couchbase_ffi/executors.py b/couchbase_ffi/executors.py index bdd43b5..e7049c5 100644 --- a/couchbase_ffi/executors.py +++ b/couchbase_ffi/executors.py @@ -589,6 +589,8 @@ 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]