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/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
@@ -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,18 +114,31 @@
 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
-    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():
     if 'eventlet.patcher' not in sys.modules:
@@ -138,40 +152,61 @@ def _is_eventlet_monkey_patched():
         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)
-
-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="AsyncoreConnection", connection_class=None, exception=e)
+
+def _try_twisted_import():
+    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)
+
+
+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_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/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

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,30 +6,10 @@
 
 from cassandra.policies import TokenAwarePolicy, RoundRobinPolicy, ConstantReconnectionPolicy
 from tests.integration import use_cluster, PROTOCOL_VERSION
-from cassandra.cluster import Cluster, TwistedConnection
+from cassandra.cluster import Cluster, get_all_supported_connections_classes
 
+supported_connection_classes = get_all_supported_connections_classes()
 
-supported_connection_classes = [TwistedConnection]
-
-try:
-    from cassandra.io.libevreactor import LibevConnection
-    supported_connection_classes += [LibevConnection]
-except ImportError:
-    pass
-
-
-try:
-    from cassandra.io.asyncorereactor import AsyncoreConnection
-    supported_connection_classes += [AsyncoreConnection]
-except ImportError:
-    pass
-
-#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__)