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

Author: absurdfarce

benchmarks/base.py

@@ -54,7 +54,7 @@
     from cassandra.io.libevreactor import LibevConnection
     have_libev = True
     supported_reactors.append(LibevConnection)
-except ImportError as exc:
+except cassandra.DependencyException as exc:
     pass
 
 have_asyncio = False

cassandra/__init__.py

@@ -735,6 +735,7 @@ class OperationType(Enum):
     Read = 0
     Write = 1
 
+
 class RateLimitReached(ConfigurationException):
     '''
     Rate limit was exceeded for a partition affected by the request.
@@ -747,3 +748,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,19 @@
 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,32 +137,46 @@ 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
-    try:
-        import eventlet.patcher
-        return eventlet.patcher.is_monkey_patched('socket')
-    # Another case related to PYTHON-1364
-    except AttributeError:
-        return False
+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
+        return (LibevConnection,None)
+    except DependencyException as e:
+        return (None, e)
 
-# 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_asyncore_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.asyncorereactor import AsyncoreConnection
+        return (AsyncoreConnection,None)
+    except DependencyException as e:
+        return (None, e)
+
+def _try_asyncio_import():
+    from cassandra.io.asyncioreactor import AsyncioConnection
+    return (AsyncioConnection, None)
+
+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, _try_asyncio_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/long/test_ipv6.py

@@ -15,19 +15,20 @@
 import os, socket, errno
 from ccmlib import common
 
+from cassandra import DependencyException
 from cassandra.cluster import NoHostAvailable
 
 try:
     from cassandra.io.asyncorereactor import AsyncoreConnection
-except ImportError:
+except DependencyException:
     AsyncoreConnection = None
 
 from tests import is_monkey_patched
 from tests.integration import use_cluster, remove_cluster, TestCluster
 
 try:
     from cassandra.io.libevreactor import LibevConnection
-except ImportError:
+except DependencyException:
     LibevConnection = None
 
 

tests/integration/standard/test_connection.py

@@ -23,27 +23,26 @@
 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 +446,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/integration/standard/test_scylla_cloud.py

@@ -5,6 +5,7 @@
 from ccmlib.utils.ssl_utils import generate_ssl_stores
 from ccmlib.utils.sni_proxy import refresh_certs, start_sni_proxy, create_cloud_config, NodeInfo
 
+from cassandra import DependencyException
 from cassandra.policies import TokenAwarePolicy, RoundRobinPolicy, ConstantReconnectionPolicy
 from tests.integration import use_cluster, PROTOCOL_VERSION
 from cassandra.cluster import Cluster, TwistedConnection
@@ -15,14 +16,14 @@
 try:
     from cassandra.io.libevreactor import LibevConnection
     supported_connection_classes += [LibevConnection]
-except ImportError:
+except DependencyException:
     pass
 
 
 try:
     from cassandra.io.asyncorereactor import AsyncoreConnection
     supported_connection_classes += [AsyncoreConnection]
-except ImportError:
+except DependencyException:
     pass
 
 #from cassandra.io.geventreactor import GeventConnection

tests/unit/io/test_asyncorereactor.py

@@ -15,11 +15,14 @@
 
 from mock import patch
 import socket
+
+from cassandra import DependencyException
+
 try:
     import cassandra.io.asyncorereactor as asyncorereactor
     from cassandra.io.asyncorereactor import AsyncoreConnection
     ASYNCCORE_AVAILABLE = True
-except ImportError:
+except (ImportError, DependencyException):
     ASYNCCORE_AVAILABLE = False
     AsyncoreConnection = None