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

Author: absurdfarce

benchmarks/base.py

@@ -28,8 +28,7 @@
 sys.path.append(os.path.join(dirname, '..'))
 
 import cassandra
-from cassandra.cluster import Cluster
-from cassandra.io.asyncorereactor import AsyncoreConnection
+from cassandra.cluster import Cluster, get_all_supported_connections_classes
 
 log = logging.getLogger()
 handler = logging.StreamHandler()
@@ -48,31 +47,7 @@
     'NOTSET': logging.NOTSET,
 }
 
-have_libev = False
-supported_reactors = [AsyncoreConnection]
-try:
-    from cassandra.io.libevreactor import LibevConnection
-    have_libev = True
-    supported_reactors.append(LibevConnection)
-except ImportError as exc:
-    pass
-
-have_asyncio = False
-try:
-    from cassandra.io.asyncioreactor import AsyncioConnection
-    have_asyncio = True
-    supported_reactors.append(AsyncioConnection)
-except (ImportError, SyntaxError):
-    pass
-
-have_twisted = False
-try:
-    from cassandra.io.twistedreactor import TwistedConnection
-    have_twisted = True
-    supported_reactors.append(TwistedConnection)
-except ImportError as exc:
-    log.exception("Error importing twisted")
-    pass
+supported_reactors = get_all_supported_connections_classes()
 
 KEYSPACE = "testkeyspace" + str(int(time.time()))
 TABLE = "testtable"
@@ -214,6 +189,15 @@ def benchmark(thread_class):
             log.info("  99.9th: %0.4fs", request_timer['999percentile'])
 
 
+def get_connection_class(class_name):
+    for cls in supported_reactors:
+        if cls.__name__ == class_name:
+            return cls
+    else:
+        log.error("unavailable reactor class: %s", class_name)
+        sys.exit(f"{class_name} is not available")
+
+
 def parse_options():
     parser = OptionParser()
     parser.add_option('-H', '--hosts', default='127.0.0.1',
@@ -261,23 +245,15 @@ def parse_options():
         log.warning("Unknown log level specified: %s; specify one of %s", options.log_level, _log_levels.keys())
 
     if options.asyncore_only:
-        options.supported_reactors = [AsyncoreConnection]
+        options.supported_reactors = [get_connection_class("AsyncoreConnection")]
     elif options.asyncio_only:
-        options.supported_reactors = [AsyncioConnection]
+        options.supported_reactors = [get_connection_class("AsyncioConnection")]
     elif options.libev_only:
-        if not have_libev:
-            log.error("libev is not available")
-            sys.exit(1)
-        options.supported_reactors = [LibevConnection]
+        options.supported_reactors = [get_connection_class("LibevConnection")]
     elif options.twisted_only:
-        if not have_twisted:
-            log.error("Twisted is not available")
-            sys.exit(1)
-        options.supported_reactors = [TwistedConnection]
+        options.supported_reactors = [get_connection_class("TwistedConnection")]
     else:
         options.supported_reactors = supported_reactors
-        if not have_libev:
-            log.warning("Not benchmarking libev reactor because libev is not available")
 
     return options, args
 

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,10 +25,11 @@
 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
+from typing import NamedTuple, Type, Optional
 from warnings import warn
 from random import random
 import re
@@ -45,12 +46,12 @@
 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,
                                   EndPoint, DefaultEndPoint, DefaultEndPointFactory,
-                                  ContinuousPagingState, SniEndPointFactory, ConnectionBusy)
+                                  ContinuousPagingState, SniEndPointFactory, ConnectionBusy, Connection)
 from cassandra.cqltypes import UserType
 import cassandra.cqltypes as types
 from cassandra.encoder import Encoder
@@ -98,12 +99,12 @@
 
 try:
     from cassandra.io.twistedreactor import TwistedConnection
-except ImportError:
+except DependencyException:
     TwistedConnection = None
 
 try:
     from cassandra.io.eventletreactor import EventletConnection
-except (ImportError, AttributeError):
+except DependencyException:
     # AttributeError was add for handling python 3.12 https://github.com/eventlet/eventlet/issues/812
     # TODO: remove it when eventlet issue would be fixed
     EventletConnection = None
@@ -113,6 +114,33 @@
 except ImportError:
     from cassandra.util import WeakSet  # NOQA
 
+
+class ClassImportResult(NamedTuple):
+    name: str
+    exception: Optional[Exception]
+    connection_class: Optional[Type[Connection]]
+
+
+def _is_gevent_monkey_patched():
+    if 'gevent.monkey' not in sys.modules:
+        return False
+    try:
+        import gevent.socket
+        return socket.socket is gevent.socket.socket
+    except (ModuleNotFoundError, ImportError, AttributeError):
+        return False
+
+def _try_gevent_import():
+    if _is_gevent_monkey_patched():
+        try:
+            from cassandra.io.geventreactor import GeventConnection
+            return ClassImportResult(name="GeventConnection", connection_class=GeventConnection, exception=None)
+        except DependencyException as e:
+            return ClassImportResult(name="GeventConnection", connection_class=None, exception=e)
+    else:
+        return ClassImportResult(name="GeventConnection", connection_class=None, exception=DependencyException("gevent is not patched"))
+
+
 def _is_eventlet_monkey_patched():
     if 'eventlet.patcher' not in sys.modules:
         return False
@@ -124,32 +152,63 @@ 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
+def _try_eventlet_import():
     try:
-        import eventlet.patcher
-        return eventlet.patcher.is_monkey_patched('socket')
-    # Another case related to PYTHON-1364
-    except AttributeError:
-        return False
+        from cassandra.io.eventletreactor import EventletConnection
+    except DependencyException as e:
+        return ClassImportResult(name="EventletConnection", connection_class=None, exception=e)
+    if _is_eventlet_monkey_patched():
+        return ClassImportResult(name="EventletConnection", connection_class=EventletConnection, exception=None)
+    return ClassImportResult(name="EventletConnection", connection_class=None, exception=DependencyException("eventlet is not patched"))
+
+def _try_libev_import():
+    try:
+        from cassandra.io.libevreactor import LibevConnection
+        return ClassImportResult(name="LibevConnection", connection_class=LibevConnection, exception=None)
+    except DependencyException as e:
+        return ClassImportResult(name="LibevConnection", connection_class=None, exception=e)
 
+def _try_asyncore_import():
+    try:
+        from cassandra.io.asyncorereactor import AsyncoreConnection
+        return ClassImportResult(name="AsyncoreConnection", connection_class=AsyncoreConnection, exception=None)
+    except DependencyException as e:
+        return ClassImportResult(name="AsyncoreConnection", connection_class=None, exception=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_twisted_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.twistedreactor import TwistedConnection
+        return ClassImportResult(name="TwistedConnection", connection_class=TwistedConnection, exception=None)
+    except DependencyException as e:
+        return ClassImportResult(name="TwistedConnection", connection_class=None, exception=e)
+
+
+log = logging.getLogger(__name__)
+
+
+def load_all_connections_classes():
+    results = []
+    for try_fn in (_try_gevent_import, _try_eventlet_import, _try_libev_import, _try_asyncore_import, _try_twisted_import):
+        results.append(try_fn())
+    return tuple(results)
+
+def get_all_supported_connections_classes():
+    return [res.connection_class for res in load_all_connections_classes() if res.connection_class]
+
+def get_default_connection_class():
+    excs = []
+    for try_fn in (_try_gevent_import, _try_eventlet_import, _try_libev_import, _try_asyncore_import, _try_twisted_import):
+        res = try_fn()
+        if res.connection_class:
+            return res.connection_class, excs
+        excs.append(res.exception)
+    return None, tuple(excs)
+
+
+(conn_class, excs) = get_default_connection_class()
+if not conn_class:
+    raise DependencyException("Unable to load a default connection class", 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/eventletreactor.py

@@ -15,10 +15,20 @@
 
 # Originally derived from MagnetoDB source:
 #   https://github.com/stackforge/magnetodb/blob/2015.1.0b1/magnetodb/common/cassandra/io/eventletreactor.py
-import eventlet
-from eventlet.green import socket
-from eventlet.queue import Queue
-from greenlet import GreenletExit
+from cassandra import DependencyException
+
+try:
+    import eventlet
+    from eventlet.green import socket
+    from eventlet.queue import Queue
+except (ModuleNotFoundError, ImportError, AttributeError):
+    raise DependencyException("Unable to import eventlet module. Try to install it via `pip install eventlet`")
+
+try:
+    from greenlet import GreenletExit
+except (ModuleNotFoundError, ImportError, AttributeError):
+    raise DependencyException("Unable to import greenlet module. Try to install it via `pip install greenlet`")
+
 import logging
 from threading import Event
 import time

cassandra/io/geventreactor.py

@@ -11,11 +11,19 @@
 # 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.
-import gevent
-import gevent.event
-from gevent.queue import Queue
-from gevent import socket
-import gevent.ssl
+from cassandra import DependencyException
+
+try:
+    import gevent
+    import gevent.event
+    from gevent.queue import Queue
+    from gevent import socket
+    import gevent.ssl
+except (ImportError, ModuleNotFoundError, AttributeError):
+    raise DependencyException(
+        "Unable to import gevent module. This module is optional, but if you want to use GeventConnection you need to "
+        "install it. Try to install it via `pip install gevent`."
+    )
 
 import logging
 import time

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 "

cassandra/io/twistedreactor.py

@@ -22,11 +22,19 @@
 from threading import Thread, Lock
 import weakref
 
-from twisted.internet import reactor, protocol
-from twisted.internet.endpoints import connectProtocol, TCP4ClientEndpoint, SSL4ClientEndpoint
-from twisted.internet.interfaces import IOpenSSLClientConnectionCreator
-from twisted.python.failure import Failure
-from zope.interface import implementer
+from cassandra import DependencyException
+try:
+    from twisted.internet import reactor, protocol
+    from twisted.internet.endpoints import connectProtocol, TCP4ClientEndpoint, SSL4ClientEndpoint
+    from twisted.internet.interfaces import IOpenSSLClientConnectionCreator
+    from twisted.python.failure import Failure
+except (ModuleNotFoundError, ImportError):
+    raise DependencyException("Unable to import twisted module. Try to install it via `pip install twisted[tls]`")
+
+try:
+    from zope.interface import implementer
+except (ModuleNotFoundError, ImportError):
+    raise DependencyException("Unable to import zope module. Try to install it via `pip install zope`")
 
 from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager, ConnectionException