PYCBC-1700: Add support for Python 3.14

Motivation
==========
Provide good user experience for those that want to use the latest
Python version, 3.14.

Modification
============
* Updated get_event_loop() utility to handle Python 3.14 behavior if the
  loop does not exist
* Moved loop validation in the async client adapter to occur prior to
  the first asynchronous operation (previously it was performed in the
  synchronous __init__())
* Added couchbase_loop_factory to enable users (Python >= 3.11) to use
  `asyncio.run(main(), loop_factory=couchbase_loop_factory)`
* Updated transcoder tests in acouchbase/txcouchbase to close newly
  created connections
* Fixed KeyValueOpTranscoder tests to try and decode in order to raise
  the expected exception (change from PYCBC-1725)
* [many|musl]linux Docker images have been updated to handle cp314
  (PYCBC-1724)

Results
=======
All builds are successful and all tests pass.

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

Author: thejcfactor

acouchbase/__init__.py

@@ -15,18 +15,29 @@
 
 import asyncio
 import selectors
+from typing import Optional
 
 
 class LoopValidator:
     REQUIRED_METHODS = {'add_reader', 'remove_reader',
                         'add_writer', 'remove_writer'}
 
     @staticmethod
-    def _get_working_loop():
-        evloop = asyncio.get_event_loop()
+    def _get_working_loop() -> asyncio.AbstractEventLoop:
+        try:
+            # Python <= 3.13: Returns existing or auto-creates a new loop.
+            # Python 3.14+: Returns existing or raises RuntimeError.
+            evloop = asyncio.get_event_loop()
+        except RuntimeError:
+            # No event loop exists in the current thread.
+            evloop = None
+
         gen_new_loop = not LoopValidator._is_valid_loop(evloop)
         if gen_new_loop:
-            evloop.close()
+            # Only attempt to close the loop if it actually exists
+            if evloop is not None and not evloop.is_closed():
+                evloop.close()
+
             selector = selectors.SelectSelector()
             new_loop = asyncio.SelectorEventLoop(selector)
             asyncio.set_event_loop(new_loop)
@@ -35,7 +46,7 @@ def _get_working_loop():
         return evloop
 
     @staticmethod
-    def _is_valid_loop(evloop):
+    def _is_valid_loop(evloop: asyncio.AbstractEventLoop) -> bool:
         if not evloop:
             return False
         for meth in LoopValidator.REQUIRED_METHODS:
@@ -46,28 +57,44 @@ def _is_valid_loop(evloop):
         return True
 
     @staticmethod
-    def get_event_loop(evloop):
+    def get_event_loop(evloop: asyncio.AbstractEventLoop) -> asyncio.AbstractEventLoop:
         if LoopValidator._is_valid_loop(evloop):
             return evloop
         return LoopValidator._get_working_loop()
 
     @staticmethod
-    def close_loop():
-        evloop = asyncio.get_event_loop()
-        evloop.close()
+    def close_loop() -> None:
+        try:
+            evloop = asyncio.get_event_loop()
+            if not evloop.is_closed():
+                evloop.close()
+        except RuntimeError:
+            # If there is no loop to get, there is no loop to close.
+            pass  # nosec
 
 
-def get_event_loop(
-    evloop=None,  # type: asyncio.AbstractEventLoop
-):
+def couchbase_loop_factory() -> asyncio.AbstractEventLoop:
     """
-    Get an event loop compatible with acouchbase.
-    Some Event loops, such as ProactorEventLoop (the default asyncio event
-    loop for Python 3.8 on Windows) are not compatible with acouchbase as
-    they don't implement all members in the abstract base class.
+    Dumb factory that simply builds a SelectorEventLoop.
+
+    Returns:
+        A SelectorEventLoop instance.
+    """
+    selector = selectors.SelectSelector()
+    return asyncio.SelectorEventLoop(selector)
 
-    :param evloop: preferred event loop
-    :return: The preferred event loop, if compatible, otherwise, a compatible
-    alternative event loop.
+
+def get_event_loop(evloop: Optional[asyncio.AbstractEventLoop] = None) -> asyncio.AbstractEventLoop:
     """
+    Get an event loop compatible with acouchbase.
+
+    Some Event loops, such as ProactorEventLoop (the default asyncio event loop for Python 3.8 on Windows)
+    are not compatible with acouchbase as they don't implement all members in the abstract base class.
+
+    Args:
+        evloop (asyncio.AbstractEventLoop, optional): An optional event loop to validate. If not provided, the default event loop will be used.
+
+    Returns:
+        The preferred event loop, if compatible, otherwise, a compatible alternative event loop.
+    """  # noqa: E501
     return LoopValidator.get_event_loop(evloop)

acouchbase/logic/client_adapter.py

@@ -47,16 +47,16 @@
 class AsyncClientAdapter:
 
     def __init__(self,
-                 loop: AbstractEventLoop,
                  connect_req: CreateConnectionRequest,
+                 loop: Optional[AbstractEventLoop] = None,
                  loop_validator: Optional[Callable[[Optional[AbstractEventLoop]], AbstractEventLoop]] = None
                  ) -> 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)
+        # The loop will be setup prior to running the first async op.
+        # This is the preferred mechanism to handling the event loop.
+        self._loop: Optional[AbstractEventLoop] = loop
+        self._loop_validator = loop_validator
         self._connect_req = connect_req
         self._binding_map = BindingMap(self._connection)
         self._close_ft: Optional[Future[None]] = None
@@ -78,6 +78,8 @@ def connect_ft(self) -> Optional[Future[None]]:
 
     @property
     def loop(self) -> AbstractEventLoop:
+        if not self._loop:
+            self._loop = self._get_loop()
         return self._loop
 
     def _ensure_not_closed(self) -> None:
@@ -101,14 +103,14 @@ def execute_bucket_request(self, req: BucketRequest) -> Future[Any]:
         self._ensure_not_closed()
         self._ensure_connected()
 
-        ft = self._loop.create_future()
+        ft = self.loop.create_future()
 
         def _callback(result) -> None:
-            self._loop.call_soon_threadsafe(ft.set_result, result)
+            self.loop.call_soon_threadsafe(ft.set_result, result)
 
         def _errback(exc) -> None:
             excptn = ErrorMapper.build_exception(exc)
-            self._loop.call_soon_threadsafe(ft.set_exception, excptn)
+            self.loop.call_soon_threadsafe(ft.set_exception, excptn)
 
         req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         self._execute_req(ft, req.op_name, req_dict)
@@ -121,14 +123,14 @@ def execute_close_bucket_request(self, bucket_name: str) -> Future[None]:
     def execute_cluster_request(self, req: ClusterRequest) -> Future[Any]:
         self._ensure_not_closed()
 
-        ft = self._loop.create_future()
+        ft = self.loop.create_future()
 
         def _callback(result) -> None:
-            self._loop.call_soon_threadsafe(ft.set_result, result)
+            self.loop.call_soon_threadsafe(ft.set_result, result)
 
         def _errback(exc) -> None:
             excptn = ErrorMapper.build_exception(exc)
-            self._loop.call_soon_threadsafe(ft.set_exception, excptn)
+            self.loop.call_soon_threadsafe(ft.set_exception, excptn)
 
         req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         if not self.connected:
@@ -150,14 +152,14 @@ def execute_collection_request(self, req: CollectionRequest) -> Future[Any]:
         self._ensure_not_closed()
         self._ensure_connected()
 
-        ft = self._loop.create_future()
+        ft = self.loop.create_future()
 
         def _callback(result) -> None:
-            self._loop.call_soon_threadsafe(ft.set_result, result)
+            self.loop.call_soon_threadsafe(ft.set_result, result)
 
         def _errback(exc) -> None:
             excptn = ErrorMapper.build_exception(exc)
-            self._loop.call_soon_threadsafe(ft.set_exception, excptn)
+            self.loop.call_soon_threadsafe(ft.set_exception, excptn)
 
         req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         self._execute_req(ft, req.op_name, req_dict)
@@ -167,16 +169,16 @@ def execute_connect_bucket_request(self, bucket_name: str) -> Future[None]:
         self._ensure_not_closed()
 
         req = OpenBucketRequest(bucket_name)
-        ft = self._loop.create_future()
+        ft = self.loop.create_future()
 
         def _callback(_) -> None:
             if not ft.done():
-                self._loop.call_soon_threadsafe(ft.set_result, None)
+                self.loop.call_soon_threadsafe(ft.set_result, None)
 
         def _errback(ret: Any) -> None:
             excptn = ErrorMapper.build_exception(ret)
             if not ft.done():
-                self._loop.call_soon_threadsafe(ft.set_exception, excptn)
+                self.loop.call_soon_threadsafe(ft.set_exception, excptn)
 
         req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         if not self.connected:
@@ -189,16 +191,16 @@ def _errback(ret: Any) -> None:
     def execute_mgmt_request(self, req: MgmtRequest) -> Future[Any]:
         self._ensure_not_closed()
 
-        ft = self._loop.create_future()
+        ft = self.loop.create_future()
 
         def _callback(ret: Any) -> None:
             if not ft.done():
-                self._loop.call_soon_threadsafe(ft.set_result, ret)
+                self.loop.call_soon_threadsafe(ft.set_result, ret)
 
         def _errback(ret: Any) -> None:
             excptn = ErrorMapper.build_exception(ret, mapping=req.error_map)
             if not ft.done():
-                self._loop.call_soon_threadsafe(ft.set_exception, excptn)
+                self.loop.call_soon_threadsafe(ft.set_exception, excptn)
 
         req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         if not self.connected:
@@ -244,17 +246,17 @@ def _execute_chained_req(self,
 
     def _execute_close_connection_request(self) -> Future[None]:
         req = CloseConnectionRequest()
-        ft = self._loop.create_future()
+        ft = self.loop.create_future()
 
         def _callback(_) -> None:
             if not ft.done():
                 self._reset_connection()
-                self._loop.call_soon_threadsafe(ft.set_result, None)
+                self.loop.call_soon_threadsafe(ft.set_result, None)
 
         def _errback(ret: Any) -> None:
             excptn = ErrorMapper.build_exception(ret)
             if not ft.done():
-                self._loop.call_soon_threadsafe(ft.set_exception, excptn)
+                self.loop.call_soon_threadsafe(ft.set_exception, excptn)
 
         req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         if not self.connected:
@@ -270,16 +272,16 @@ def _errback(ret: Any) -> None:
         return ft
 
     def _execute_connect_request(self) -> Future[None]:
-        ft = self._loop.create_future()
+        ft = self.loop.create_future()
 
         def _callback(_) -> None:
             if not ft.done():
-                self._loop.call_soon_threadsafe(ft.set_result, None)
+                self.loop.call_soon_threadsafe(ft.set_result, None)
 
         def _errback(ret: Any) -> None:
             excptn = ErrorMapper.build_exception(ret)
             if not ft.done():
-                self._loop.call_soon_threadsafe(ft.set_exception, excptn)
+                self.loop.call_soon_threadsafe(ft.set_exception, excptn)
 
         req_dict = self._connect_req.req_to_dict(callback=_callback, errback=_errback)
         self._execute_req(ft, self._connect_req.op_name, req_dict)
@@ -316,7 +318,10 @@ def _execute_req_sync(self, op_name: str, req_dict: Dict[str, Any]) -> Any:
 
     def _get_loop(self, loop: Optional[AbstractEventLoop] = None) -> AbstractEventLoop:
         if not loop:
-            loop = get_event_loop()
+            if self._loop_validator:
+                loop = self._loop_validator(loop)
+            else:
+                loop = get_event_loop()
 
         if not loop.is_running():
             raise RuntimeError('Event loop is not running.')

acouchbase/logic/cluster_impl.py

@@ -53,7 +53,7 @@
 
 class AsyncClusterImpl:
     def __init__(self, connstr: str, *options: object, **kwargs: object) -> None:
-        loop = kwargs.pop('loop', None)
+        loop: Optional[AbstractEventLoop] = kwargs.pop('loop', None)
         loop_validator = kwargs.pop('loop_validator', None)
         self._cluster_settings = ClusterSettings.build_cluster_settings(connstr, *options, **kwargs)
         connect_request = CreateConnectionRequest(self._cluster_settings.connstr,
@@ -62,7 +62,7 @@ def __init__(self, connstr: str, *options: object, **kwargs: object) -> None:
         # A connection is made when we create the client adapter, but it is an async operation that we cannot await
         # b/c the call needs to happen when we initialize a cluster (new cluster -> new client adapter). We await
         # the create connection future in whichever operation comes next.
-        self._client_adapter = AsyncClientAdapter(loop, connect_request, loop_validator=loop_validator)
+        self._client_adapter = AsyncClientAdapter(connect_request, loop=loop, loop_validator=loop_validator)
         self._request_builder = ClusterRequestBuilder()
         self._cluster_info: Optional[ClusterInfoResult] = None
         self._transactions: Optional[Transactions] = None