fix(cassandra.io): fix all problems around DependencyException

Upstream introduced `DependencyException` and fixed some of the
connection classes to raise it when required module is absent.
It was done half-way and broke lots of tests.
This commit fixes rest of the code and problems original commit created.

Author: dkropachev

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/cluster.py

@@ -29,6 +29,7 @@
 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
@@ -50,7 +51,7 @@
 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
@@ -103,7 +104,7 @@
 
 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
@@ -114,19 +115,30 @@
     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
-    import gevent.socket
-    return socket.socket is gevent.socket.socket
-
+    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():
-        from cassandra.io.geventreactor import GeventConnection
-        return (GeventConnection,None)
+        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 (None,None)
+        return ClassImportResult(name="GeventConnection", connection_class=None, exception=DependencyException("gevent is not patched"))
 
 
 def _is_eventlet_monkey_patched():
@@ -140,77 +152,56 @@ 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():
+    try:
         from cassandra.io.eventletreactor import EventletConnection
-        return (EventletConnection,None)
-    else:
-        return (None,None)
+    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 (LibevConnection,None)
+        return ClassImportResult(name="LibevConnection", connection_class=LibevConnection, exception=None)
     except DependencyException as e:
-        return (None, e)
+        return ClassImportResult(name="LibevConnection", connection_class=None, exception=e)
 
 def _try_asyncore_import():
     try:
         from cassandra.io.asyncorereactor import AsyncoreConnection
-        return (AsyncoreConnection,None)
+        return ClassImportResult(name="AsyncoreConnection", connection_class=AsyncoreConnection, exception=None)
     except DependencyException as e:
-        return (None, e)
+        return ClassImportResult(name="AsyncoreConnection", connection_class=None, exception=e)
 
 def _try_twisted_import():
     try:
         from cassandra.io.twistedreactor import TwistedConnection
-        return TwistedConnection, None
+        return ClassImportResult(name="TwistedConnection", connection_class=TwistedConnection, exception=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)
+        return ClassImportResult(name="TwistedConnection", connection_class=None, exception=e)
+
 
 log = logging.getLogger(__name__)
 
-def get_all_supported_connections_classes():
-    classes = []
-    excs = []
+
+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):
-        conn, exc = try_fn()
-        if conn is not None:
-            classes.append(conn)
-        else:
-            excs.append(exc)
-    return tuple(classes), tuple(excs)
+        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):
-        conn, exc = try_fn()
-        if conn is not None:
-            return conn, None
-        else:
-            excs.append(exc)
+        res = try_fn()
+        if res.connection_class:
+            return res.connection_class, excs
+        excs.append(res.exception)
     return None, tuple(excs)
 
 

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

tests/__init__.py

@@ -66,7 +66,7 @@ def is_monkey_patched():
         gevent.monkey.patch_all()
         from cassandra.io.geventreactor import GeventConnection
         connection_class = GeventConnection
-    except ImportError:
+    except DependencyException:
         connection_class = None
 elif "eventlet" in EVENT_LOOP_MANAGER:
     from eventlet import monkey_patch

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_scylla_cloud.py

@@ -1,4 +1,3 @@
-import json
 import logging
 import os.path
 from unittest import TestCase
@@ -7,18 +6,10 @@
 
 from cassandra.policies import TokenAwarePolicy, RoundRobinPolicy, ConstantReconnectionPolicy
 from tests.integration import use_cluster, PROTOCOL_VERSION
-from cassandra.cluster import Cluster, TwistedConnection
-from tests.integration import use_cluster
 from cassandra.cluster import Cluster, get_all_supported_connections_classes
 
-supported_connection_classes, _ = get_all_supported_connections_classes()
+supported_connection_classes = get_all_supported_connections_classes()
 
-#from cassandra.io.geventreactor import GeventConnection
-#from cassandra.io.eventletreactor import EventletConnection
-#from cassandra.io.asyncioreactor import AsyncioConnection
-
-# need to run them with specific configuration like `gevent.monkey.patch_all()` or under async functions
-# unsupported_connection_classes = [GeventConnection, AsyncioConnection, EventletConnection]
 LOGGER = logging.getLogger(__name__)
 
 

tests/unit/io/test_asyncorereactor.py

@@ -12,20 +12,22 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import unittest
+import socket
 
 from mock import patch
-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:
     import cassandra.io.asyncorereactor as asyncorereactor
     from cassandra.io.asyncorereactor import AsyncoreConnection
     ASYNCCORE_AVAILABLE = True
-except ImportError:
+except DependencyException:
     ASYNCCORE_AVAILABLE = False
     AsyncoreConnection = None
 
-from tests import is_monkey_patched
-from tests.unit.io.utils import ReactorTestMixin, TimerTestMixin, noop_if_monkey_patched
-
 
 @unittest.skipIf(not ASYNCCORE_AVAILABLE, 'asyncore is deprecated')
 class AsyncorePatcher(unittest.TestCase):