PYTHON-1366 Handle removal of asyncore in Python 3.12 (datastax#1187)

Author: absurdfarce

cassandra/__init__.py

@@ -747,3 +747,20 @@ def __init__(self, op_type=None, rejected_by_coordinator=False):
         self.rejected_by_coordinator = rejected_by_coordinator
         message = f"[request_error_rate_limit_reached OpType={op_type.name} RejectedByCoordinator={rejected_by_coordinator}]"
         Exception.__init__(self, message)
+
+
+class DependencyException(Exception):
+    """
+    Specific exception class for handling issues with driver dependencies
+    """
+
+    excs = []
+    """
+    A sequence of child exceptions
+    """
+
+    def __init__(self, msg, excs=[]):
+        complete_msg = msg
+        if excs:
+            complete_msg += ("The following exceptions were observed: \n" + '\n'.join(str(e) for e in excs))
+        Exception.__init__(self, complete_msg)

cassandra/cluster.py

@@ -25,7 +25,7 @@
 from collections.abc import Mapping
 from concurrent.futures import ThreadPoolExecutor, FIRST_COMPLETED, wait as wait_futures
 from copy import copy
-from functools import partial, wraps
+from functools import partial, reduce, wraps
 from itertools import groupby, count, chain
 import json
 import logging
@@ -45,7 +45,7 @@
 from cassandra import (ConsistencyLevel, AuthenticationFailed, InvalidRequest,
                        OperationTimedOut, UnsupportedOperation,
                        SchemaTargetType, DriverException, ProtocolVersion,
-                       UnresolvableContactPoints)
+                       UnresolvableContactPoints, DependencyException)
 from cassandra.auth import _proxy_execute_key, PlainTextAuthProvider
 from cassandra.connection import (ConnectionException, ConnectionShutdown,
                                   ConnectionHeartbeat, ProtocolVersionUnsupported,
@@ -113,6 +113,22 @@
 except ImportError:
     from cassandra.util import WeakSet  # NOQA
 
+
+def _is_gevent_monkey_patched():
+    if 'gevent.monkey' not in sys.modules:
+        return False
+    import gevent.socket
+    return socket.socket is gevent.socket.socket
+
+
+def _try_gevent_import():
+    if _is_gevent_monkey_patched():
+        from cassandra.io.geventreactor import GeventConnection
+        return (GeventConnection,None)
+    else:
+        return (None,None)
+
+
 def _is_eventlet_monkey_patched():
     if 'eventlet.patcher' not in sys.modules:
         return False
@@ -124,6 +140,7 @@ def _is_eventlet_monkey_patched():
         # TODO: remove it when eventlet issue would be fixed
         return False
 
+
 def _is_gevent_monkey_patched():
     if 'gevent.monkey' not in sys.modules:
         return False
@@ -135,21 +152,42 @@ def _is_gevent_monkey_patched():
         return False
 
 
-# default to gevent when we are monkey patched with gevent, eventlet when
-# monkey patched with eventlet, otherwise if libev is available, use that as
-# the default because it's fastest. Otherwise, use asyncore.
-if _is_gevent_monkey_patched():
-    from cassandra.io.geventreactor import GeventConnection as DefaultConnection
-elif _is_eventlet_monkey_patched():
-    from cassandra.io.eventletreactor import EventletConnection as DefaultConnection
-else:
+def _try_eventlet_import():
+    if _is_eventlet_monkey_patched():
+        from cassandra.io.eventletreactor import EventletConnection
+        return (EventletConnection,None)
+    else:
+        return (None,None)
+
+def _try_libev_import():
     try:
-        from cassandra.io.libevreactor import LibevConnection as DefaultConnection  # NOQA
-    except ImportError:
-        try:
-            from cassandra.io.asyncorereactor import AsyncoreConnection as DefaultConnection # NOQA
-        except ImportError:
-            from cassandra.io.asyncioreactor import AsyncioConnection as DefaultConnection  # NOQA
+        from cassandra.io.libevreactor import LibevConnection
+        return (LibevConnection,None)
+    except DependencyException as e:
+        return (None, e)
+
+def _try_asyncore_import():
+    try:
+        from cassandra.io.asyncorereactor import AsyncoreConnection
+        return (AsyncoreConnection,None)
+    except DependencyException as e:
+        return (None, e)
+
+def _connection_reduce_fn(val,import_fn):
+    (rv, excs) = val
+    # If we've already found a workable Connection class return immediately
+    if rv:
+        return val
+    (import_result, exc) = import_fn()
+    if exc:
+        excs.append(exc)
+    return (rv or import_result, excs)
+
+conn_fns = (_try_gevent_import, _try_eventlet_import, _try_libev_import, _try_asyncore_import)
+(conn_class, excs) = reduce(_connection_reduce_fn, conn_fns, (None,[]))
+if excs:
+    raise DependencyException("Exception loading connection class dependencies", excs)
+DefaultConnection = conn_class
 
 # Forces load of utf8 encoding module to avoid deadlock that occurs
 # if code that is being imported tries to import the module in a seperate

cassandra/io/asyncorereactor.py

@@ -30,7 +30,15 @@
 except ImportError:
     from cassandra.util import WeakSet  # noqa
 
-import asyncore
+from cassandra import DependencyException
+try:
+    import asyncore
+except ModuleNotFoundError:
+    raise DependencyException(
+        "Unable to import asyncore module.  Note that this module has been removed in Python 3.12 "
+        "so when using the driver with this version (or anything newer) you will need to use one of the "
+        "other event loop implementations."
+    )
 
 from cassandra.connection import Connection, ConnectionShutdown, NONBLOCKING, Timer, TimerManager
 

cassandra/io/libevreactor.py

@@ -21,13 +21,13 @@
 from threading import Lock, Thread
 import time
 
-
+from cassandra import DependencyException
 from cassandra.connection import (Connection, ConnectionShutdown,
                                   NONBLOCKING, Timer, TimerManager)
 try:
     import cassandra.io.libevwrapper as libev
 except ImportError:
-    raise ImportError(
+    raise DependencyException(
         "The C extension needed to use libev was not found.  This "
         "probably means that you didn't have the required build dependencies "
         "when installing the driver.  See "

tests/__init__.py

@@ -20,17 +20,19 @@
 import os
 from concurrent.futures import ThreadPoolExecutor
 
+from cassandra import DependencyException
+
 log = logging.getLogger()
 
 def is_eventlet_monkey_patched():
     if 'eventlet.patcher' not in sys.modules:
         return False
     try:
         import eventlet.patcher
+        return eventlet.patcher.is_monkey_patched('socket')
+    # Yet another case related to PYTHON-1364
     except AttributeError:
         return False
-    return eventlet.patcher.is_monkey_patched('socket')
-
 
 def is_gevent_monkey_patched():
     if 'gevent.monkey' not in sys.modules:
@@ -88,17 +90,18 @@ def is_monkey_patched():
 elif "asyncio" in EVENT_LOOP_MANAGER:
     from cassandra.io.asyncioreactor import AsyncioConnection
     connection_class = AsyncioConnection
-
 else:
+    log.debug("Using default event loop (libev)")
     try:
         from cassandra.io.libevreactor import LibevConnection
         connection_class = LibevConnection
-    except ImportError as e:
+    except DependencyException as e:
         log.debug('Could not import LibevConnection, '
                   'using connection_class=None; '
                   'failed with error:\n {}'.format(
                       repr(e)
                   ))
+        log.debug("Will attempt to set connection class at cluster initialization")
         connection_class = None
 
 

tests/integration/standard/test_connection.py

@@ -23,27 +23,27 @@
 import time
 from unittest import SkipTest
 
-from cassandra import ConsistencyLevel, OperationTimedOut
+from cassandra import ConsistencyLevel, OperationTimedOut, DependencyException
 from cassandra.cluster import NoHostAvailable, ConnectionShutdown, ExecutionProfile, EXEC_PROFILE_DEFAULT
 
-try:
-    from cassandra.io.asyncorereactor import AsyncoreConnection
-except ImportError:
-    AsyncoreConnection = None
-
 from cassandra.protocol import QueryMessage
-from cassandra.connection import Connection
 from cassandra.policies import HostFilterPolicy, RoundRobinPolicy, HostStateListener
 from cassandra.pool import HostConnectionPool
 
 from tests import is_monkey_patched
 from tests.integration import use_singledc, get_node, CASSANDRA_IP, local, \
     requiresmallclockgranularity, greaterthancass20, TestCluster
 
+try:
+    import cassandra.io.asyncorereactor
+    from cassandra.io.asyncorereactor import AsyncoreConnection
+except DependencyException:
+    AsyncoreConnection = None
+
 try:
     from cassandra.io.libevreactor import LibevConnection
     import cassandra.io.libevreactor
-except ImportError:
+except DependencyException:
     LibevConnection = None
 
 
@@ -447,8 +447,7 @@ def setUp(self):
         if is_monkey_patched():
             raise unittest.SkipTest("Can't test asyncore with monkey patching")
         if AsyncoreConnection is None:
-            raise unittest.SkipTest(
-                'asyncore does not appear to be installed properly')
+            raise unittest.SkipTest('Unable to import asyncore module')
         ConnectionTests.setUp(self)
 
     def clean_global_loop(self):

tests/unit/io/test_libevreactor.py

@@ -14,17 +14,17 @@
 import unittest
 
 from mock import patch, Mock
-import weakref
 import socket
 
+from cassandra import DependencyException
 from tests import is_monkey_patched
 from tests.unit.io.utils import ReactorTestMixin, TimerTestMixin, noop_if_monkey_patched
 
 
 try:
     from cassandra.io.libevreactor import _cleanup as libev__cleanup
     from cassandra.io.libevreactor import LibevConnection
-except ImportError:
+except DependencyException:
     LibevConnection = None  # noqa