Cleanup after PYTHON-1366

Author: absurdfarce

benchmarks/base.py

@@ -23,13 +23,14 @@
 
 from greplin import scales
 
+from cassandra.io.reactorloader import get_all_supported_connections_classes
+
 dirname = os.path.dirname(os.path.abspath(__file__))
 sys.path.append(dirname)
 sys.path.append(os.path.join(dirname, '..'))
 
 import cassandra
 from cassandra.cluster import Cluster
-from cassandra.io.asyncorereactor import AsyncoreConnection
 
 log = logging.getLogger()
 handler = logging.StreamHandler()
@@ -48,31 +49,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 +191,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 +247,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

@@ -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, reduce, wraps
+from functools import partial, wraps
 from itertools import groupby, count, chain
 import json
 import logging
@@ -50,10 +50,11 @@
 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
+from cassandra.io.reactorloader import get_default_connection_class, try_twisted_connection
 from cassandra.protocol import (QueryMessage, ResultMessage,
                                 ErrorMessage, ReadTimeoutErrorMessage,
                                 WriteTimeoutErrorMessage,
@@ -96,82 +97,19 @@
 from cassandra.datastax import cloud as dscloud
 from cassandra.scylla.cloud import CloudConfiguration
 
-try:
-    from cassandra.io.twistedreactor import TwistedConnection
-except ImportError:
-    TwistedConnection = None
-
-try:
-    from cassandra.io.eventletreactor import EventletConnection
-except (ImportError, AttributeError):
-    # 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
+TwistedConnection = try_twisted_connection().connection_class
+EventletConnection = try_twisted_connection().connection_class
 
 try:
     from weakref import WeakSet
 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
-    try:
-        import eventlet.patcher
-        return eventlet.patcher.is_monkey_patched('socket')
-    except (ImportError, AttributeError):
-        # AttributeError was add for handling python 3.12 https://github.com/eventlet/eventlet/issues/812
-        # TODO: remove it when eventlet issue would be fixed
-        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)
-
-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)
+log = logging.getLogger(__name__)
 
-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)
+(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

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

@@ -0,0 +1,100 @@
+import sys
+from socket import socket
+from typing import NamedTuple, Optional, Type
+
+from cassandra import DependencyException
+from cassandra.connection import Connection
+
+
+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_connection():
+    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
+    try:
+        import eventlet.patcher
+        return eventlet.patcher.is_monkey_patched('socket')
+    except (ImportError, AttributeError):
+        # AttributeError was add for handling python 3.12 https://github.com/eventlet/eventlet/issues/812
+        # TODO: remove it when eventlet issue would be fixed
+        return False
+
+
+def try_eventlet_connection():
+    try:
+        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_connection():
+    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)
+
+
+def try_twisted_connection():
+    try:
+        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)
+
+
+def load_all_connections_classes():
+    results = []
+    for try_fn in (try_gevent_connection, try_eventlet_connection, try_libev_import, try_asyncore_connection, try_twisted_connection):
+        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_connection, try_eventlet_connection, try_libev_import, try_asyncore_connection, try_twisted_connection):
+        res = try_fn()
+        if res.connection_class:
+            return res.connection_class, excs
+        excs.append(res.exception)
+    return None, tuple(excs)

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
 

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