PYCBC-1745: Use C++ Templating System for SDK C-Extension

Motivation
----------
Moving the client's c-extension to utilize C++ templating will help prepare for future
changes and continue to improve the maintainability of the Couchbase Python client.

Changes
-------
* Use C++ templates to reduce the execution paths so maintenance becomes more straight forward (e.g. 50+ paths to ~10)
* Use C++ templates to better map between Python and C++ objects
* Namespace the code in the extension to pcybc
* Clean up transactions logic
    - New new template system makes for easier creation of config/options objects
    - Cleaned up how we add txns objects (PYCBC-1686)
* Clean up exception handling
* Help to set the stage for the SDK to fully support static typing
* Improved handling of the couchbase API’s multi operations
    - Added tests to confirm durability options cannot be mixed
* Move c-extension shared object to reside in couchbase/logic/pycbc_core
  to help with static typing and imports

Change-Id: Id672f13f9273fa0c2ff1b3698f7950375a383445
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/240534
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Dimitris Christodoulou <dimitris.christodoulou@couchbase.com>

Author: thejcfactor

CMakeLists.txt

@@ -249,7 +249,6 @@ file(
   GLOB
   SOURCE_FILES
   "src/*.cxx"
-  "src/management/*.cxx"
   "src/transactions/*.cxx")
 add_library(pycbc_core SHARED ${SOURCE_FILES})
 
@@ -301,5 +300,5 @@ endif()
 set_target_properties(
   pycbc_core
   PROPERTIES PREFIX ""
-             OUTPUT_NAME pycbc_core
+             OUTPUT_NAME _core
              SUFFIX ${PYCBC_C_MOD_SUFFIX})

acouchbase/analytics.py

@@ -22,9 +22,9 @@
                                   CouchbaseException,
                                   ErrorMapper,
                                   ExceptionMap)
-from couchbase.exceptions import exception as CouchbaseBaseException
 from couchbase.logic.analytics import AnalyticsQuery  # noqa: F401
 from couchbase.logic.analytics import AnalyticsRequestLogic
+from couchbase.logic.pycbc_core import pycbc_exception as PycbcCoreException
 
 
 class AsyncAnalyticsRequest(AnalyticsRequestLogic):
@@ -82,7 +82,7 @@ def _get_next_row(self):
             return
 
         row = next(self._streaming_result)
-        if isinstance(row, CouchbaseBaseException):
+        if isinstance(row, PycbcCoreException):
             raise ErrorMapper.build_exception(row)
         # should only be None one query request is complete and _no_ errors found
         if row is None:

acouchbase/collection.py

@@ -1030,7 +1030,8 @@ def scan(self,
 
 
         """  # noqa: E501
-        req = self._impl.request_builder.build_range_scan_async_request(scan_type, *opts, **kwargs)
+        req = self._impl.request_builder.build_range_scan_async_request(
+            self._impl.connection, scan_type, *opts, **kwargs)
         return self._impl.range_scan(req)
 
     def binary(self) -> BinaryCollection:

acouchbase/kv_range_scan.py

@@ -12,10 +12,12 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-#
+
+from __future__ import annotations
 
 import asyncio
 from concurrent.futures import ThreadPoolExecutor
+from typing import TYPE_CHECKING, Any
 
 from couchbase.exceptions import (PYCBC_ERROR_MAP,
                                   AlreadyQueriedException,
@@ -24,11 +26,14 @@
                                   RangeScanCompletedException)
 from couchbase.logic.kv_range_scan import RangeScanRequestLogic
 
+if TYPE_CHECKING:
+    from couchbase.logic.pycbc_core import pycbc_connection
+
 
 class AsyncRangeScanRequest(RangeScanRequestLogic):
-    def __init__(self, loop: asyncio.AbstractEventLoop, **kwargs: object) -> None:
+    def __init__(self, connection: pycbc_connection, loop: asyncio.AbstractEventLoop, **kwargs: Any) -> None:
         num_workers = kwargs.pop('num_workers', 2)
-        super().__init__(**kwargs)
+        super().__init__(connection, **kwargs)
         self._loop = loop
         self._result_ftr = None
         self._tp_executor = ThreadPoolExecutor(num_workers)

acouchbase/logic/bucket_impl.py

@@ -16,14 +16,12 @@
 from __future__ import annotations
 
 from asyncio import AbstractEventLoop, Future
-from typing import (TYPE_CHECKING,
-                    Optional,
-                    Union)
+from typing import TYPE_CHECKING, Union
 
 from acouchbase.views import AsyncViewRequest
 from couchbase.logic.bucket_req_builder import BucketRequestBuilder
-from couchbase.logic.client_adapter import PyCapsuleType
 from couchbase.logic.cluster_settings import ClusterSettings, StreamingTimeouts
+from couchbase.logic.pycbc_core import pycbc_connection
 from couchbase.result import PingResult, ViewResult
 
 if TYPE_CHECKING:
@@ -68,7 +66,7 @@ def connected(self) -> bool:
         return self._client_adapter.connected
 
     @property
-    def connection(self) -> Optional[PyCapsuleType]:
+    def connection(self) -> pycbc_connection:
         """**INTERNAL**"""
         return self._client_adapter.connection
 

acouchbase/logic/client_adapter.py

@@ -29,11 +29,11 @@
                                   ErrorMapper,
                                   ExceptionMap,
                                   InternalSDKException)
-from couchbase.exceptions import exception as BaseCouchbaseException
 from couchbase.logic.binding_map import BindingMap
 from couchbase.logic.bucket_types import CloseBucketRequest, OpenBucketRequest
 from couchbase.logic.cluster_types import CloseConnectionRequest
-from couchbase.logic.top_level_types import OpenOrCloseBucket, PyCapsuleType
+from couchbase.logic.pycbc_core import pycbc_connection
+from couchbase.logic.pycbc_core import pycbc_exception as PycbcCoreException
 
 if TYPE_CHECKING:
     from asyncio import AbstractEventLoop
@@ -51,24 +51,25 @@ def __init__(self,
                  connect_req: CreateConnectionRequest,
                  loop_validator: Optional[Callable[[Optional[AbstractEventLoop]], AbstractEventLoop]] = None
                  ) -> None:
-        self._connection: Optional[PyCapsuleType] = None
+        num_io_threads = connect_req.options.get('num_io_threads', None)
+        self._connection = pycbc_connection(num_io_threads) if num_io_threads is not None else pycbc_connection()
         if loop_validator:
             self._loop = loop_validator(loop)
         else:
             self._loop = self._get_loop(loop)
         self._connect_req = connect_req
-        self._binding_map = BindingMap()
+        self._binding_map = BindingMap(self._connection)
         self._close_ft: Optional[Future[None]] = None
         self._connect_ft: Optional[Future[None]] = None
         self._closed = False
         self._create_connection()
 
     @property
     def connected(self) -> bool:
-        return self._connection is not None
+        return self._connection is not None and self._connection.connected
 
     @property
-    def connection(self) -> Optional[PyCapsuleType]:
+    def connection(self) -> pycbc_connection:
         return self._connection
 
     @property
@@ -109,12 +110,12 @@ def _errback(exc) -> None:
             excptn = ErrorMapper.build_exception(exc)
             self._loop.call_soon_threadsafe(ft.set_exception, excptn)
 
-        req_dict = req.req_to_dict(self._connection, callback=_callback, errback=_errback)
+        req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         self._execute_req(ft, req.op_name, req_dict)
         return ft
 
     def execute_close_bucket_request(self, bucket_name: str) -> Future[None]:
-        req = CloseBucketRequest(bucket_name, OpenOrCloseBucket.CLOSE)
+        req = CloseBucketRequest(bucket_name)
         return self.execute_bucket_request(req)
 
     def execute_cluster_request(self, req: ClusterRequest) -> Future[Any]:
@@ -129,7 +130,7 @@ def _errback(exc) -> None:
             excptn = ErrorMapper.build_exception(exc)
             self._loop.call_soon_threadsafe(ft.set_exception, excptn)
 
-        req_dict = req.req_to_dict(self._connection, callback=_callback, errback=_errback)
+        req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         if not self.connected:
             chained_ft = self._execute_connect_request() if self._connect_ft is None else self._connect_ft
             chained_ft.add_done_callback(partial(self._execute_chained_req, ft, req.op_name, req_dict))
@@ -139,9 +140,9 @@ def _errback(exc) -> None:
 
     def execute_cluster_request_sync(self, req: ClusterRequest) -> Any:
         self._ensure_not_closed()
-        req_dict = req.req_to_dict(self._connection)
+        req_dict = req.req_to_dict()
         ret = self._execute_req_sync(req.op_name, req_dict)
-        if isinstance(ret, BaseCouchbaseException):
+        if isinstance(ret, PycbcCoreException):
             raise ErrorMapper.build_exception(ret)
         return ret
 
@@ -158,14 +159,14 @@ def _errback(exc) -> None:
             excptn = ErrorMapper.build_exception(exc)
             self._loop.call_soon_threadsafe(ft.set_exception, excptn)
 
-        req_dict = req.req_to_dict(self._connection, callback=_callback, errback=_errback)
+        req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         self._execute_req(ft, req.op_name, req_dict)
         return ft
 
     def execute_connect_bucket_request(self, bucket_name: str) -> Future[None]:
         self._ensure_not_closed()
 
-        req = OpenBucketRequest(bucket_name, OpenOrCloseBucket.OPEN)
+        req = OpenBucketRequest(bucket_name)
         ft = self._loop.create_future()
 
         def _callback(_) -> None:
@@ -177,7 +178,7 @@ def _errback(ret: Any) -> None:
             if not ft.done():
                 self._loop.call_soon_threadsafe(ft.set_exception, excptn)
 
-        req_dict = req.req_to_dict(self._connection, callback=_callback, errback=_errback)
+        req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         if not self.connected:
             chained_ft = self._execute_connect_request() if self._connect_ft is None else self._connect_ft
             chained_ft.add_done_callback(partial(self._execute_chained_req, ft, req.op_name, req_dict))
@@ -199,7 +200,7 @@ def _errback(ret: Any) -> None:
             if not ft.done():
                 self._loop.call_soon_threadsafe(ft.set_exception, excptn)
 
-        req_dict = req.req_to_dict(self._connection, callback=_callback, errback=_errback)
+        req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         if not self.connected:
             chained_ft = self._execute_connect_request() if self._connect_ft is None else self._connect_ft
             chained_ft.add_done_callback(partial(self._execute_chained_req, ft, req.op_name, req_dict))
@@ -239,7 +240,6 @@ def _execute_chained_req(self,
             ft.set_exception(exc)
             return
 
-        req_dict['conn'] = self._connection
         self._execute_req(ft, op_name, req_dict)
 
     def _execute_close_connection_request(self) -> Future[None]:
@@ -256,7 +256,7 @@ def _errback(ret: Any) -> None:
             if not ft.done():
                 self._loop.call_soon_threadsafe(ft.set_exception, excptn)
 
-        req_dict = req.req_to_dict(self._connection, callback=_callback, errback=_errback)
+        req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         if not self.connected:
             # If we're closed, don't try to reconnect just to close again
             if self._closed:
@@ -272,9 +272,8 @@ def _errback(ret: Any) -> None:
     def _execute_connect_request(self) -> Future[None]:
         ft = self._loop.create_future()
 
-        def _callback(ret: PyCapsuleType) -> None:
+        def _callback(_) -> None:
             if not ft.done():
-                self._connection = ret
                 self._loop.call_soon_threadsafe(ft.set_result, None)
 
         def _errback(ret: Any) -> None:

acouchbase/logic/cluster_impl.py

@@ -31,7 +31,7 @@
 from couchbase.logic.cluster_impl import ClusterSettings
 from couchbase.logic.cluster_req_builder import ClusterRequestBuilder
 from couchbase.logic.cluster_types import CreateConnectionRequest, GetConnectionInfoRequest
-from couchbase.logic.top_level_types import PyCapsuleType
+from couchbase.logic.pycbc_core import pycbc_connection
 from couchbase.result import (AnalyticsResult,
                               ClusterInfoResult,
                               DiagnosticsResult,
@@ -88,7 +88,7 @@ def connected(self) -> bool:
         return self._client_adapter.connected
 
     @property
-    def connection(self) -> Optional[PyCapsuleType]:
+    def connection(self) -> pycbc_connection:
         """**INTERNAL**"""
         return self._client_adapter.connection