CASSPYTHON-13: Remove eventlet, gevent and twisted event loops

CONTRIBUTING.rst

@@ -73,8 +73,7 @@ Running Unit Tests
 Unit tests can be run like so::
 
     uv run pytest tests/unit
-    EVENT_LOOP_MANAGER=gevent uv run pytest tests/unit/io/test_geventreactor.py
-    EVENT_LOOP_MANAGER=eventlet uv run pytest tests/unit/io/test_eventletreactor.py
+    EVENT_LOOP_MANAGER=asyncio uv run pytest tests/unit/io/test_asyncioreactor.py
 
 You can run a specific test method like so::
 

benchmarks/base.py

@@ -65,15 +65,6 @@
 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
-
 KEYSPACE = "testkeyspace" + str(int(time.time()))
 TABLE = "testtable"
 
@@ -228,8 +219,6 @@ def parse_options():
                       help='only benchmark with asyncio connections')
     parser.add_option('--libev-only', action='store_true', dest='libev_only',
                       help='only benchmark with libev connections')
-    parser.add_option('--twisted-only', action='store_true', dest='twisted_only',
-                      help='only benchmark with Twisted connections')
     parser.add_option('-m', '--metrics', action='store_true', dest='enable_metrics',
                       help='enable and print metrics for operations')
     parser.add_option('-l', '--log-level', default='info',
@@ -269,11 +258,6 @@ def parse_options():
             log.error("libev is not available")
             sys.exit(1)
         options.supported_reactors = [LibevConnection]
-    elif options.twisted_only:
-        if not have_twisted:
-            log.error("Twisted is not available")
-            sys.exit(1)
-        options.supported_reactors = [TwistedConnection]
     else:
         options.supported_reactors = supported_reactors
         if not have_libev:

cassandra/cluster.py

@@ -37,7 +37,6 @@
 import re
 import queue
 import socket
-import sys
 import time
 from threading import Lock, RLock, Thread, Event
 import uuid
@@ -99,57 +98,11 @@
 from cassandra.datastax import cloud as dscloud
 from cassandra.application_info import ApplicationInfoBase
 
-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
-
 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
-    try:
-        import gevent.socket
-        return socket.socket is gevent.socket.socket    # Another case related to PYTHON-1364
-    except (AttributeError, ImportError):
-        return False
-
-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
@@ -178,7 +131,7 @@ def _connection_reduce_fn(val,import_fn):
         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_fns = (_try_libev_import, _try_asyncore_import, _try_asyncio_import)
 (conn_class, excs) = reduce(_connection_reduce_fn, conn_fns, (None,[]))
 if not conn_class:
     raise DependencyException("Exception loading connection class dependencies", excs)
@@ -945,19 +898,13 @@ def default_retry_policy(self, policy):
 
     * :class:`cassandra.io.asyncorereactor.AsyncoreConnection`
     * :class:`cassandra.io.libevreactor.LibevConnection`
-    * :class:`cassandra.io.eventletreactor.EventletConnection` (requires monkey-patching - see doc for details)
-    * :class:`cassandra.io.geventreactor.GeventConnection` (requires monkey-patching - see doc for details)
-    * :class:`cassandra.io.twistedreactor.TwistedConnection`
     * EXPERIMENTAL: :class:`cassandra.io.asyncioreactor.AsyncioConnection`
 
     By default, ``AsyncoreConnection`` will be used, which uses
     the ``asyncore`` module in the Python standard library.
 
     If ``libev`` is installed, ``LibevConnection`` will be used instead.
 
-    If ``gevent`` or ``eventlet`` monkey-patching is detected, the corresponding
-    connection class will be used automatically.
-
     ``AsyncioConnection``, which uses the ``asyncio`` module in the Python
     standard library, is also available, but currently experimental. Note that
     it requires ``asyncio`` features that were only introduced in the 3.4 line
@@ -1302,9 +1249,7 @@ def __init__(self,
                 raise ValueError("contact_points, endpoint_factory, ssl_context, and ssl_options "
                                  "cannot be specified with a cloud configuration")
 
-            uses_twisted = TwistedConnection and issubclass(self.connection_class, TwistedConnection)
-            uses_eventlet = EventletConnection and issubclass(self.connection_class, EventletConnection)
-            cloud_config = dscloud.get_cloud_config(cloud, create_pyopenssl_context=uses_twisted or uses_eventlet)
+            cloud_config = dscloud.get_cloud_config(cloud)
 
             ssl_context = cloud_config.ssl_context
             ssl_options = {'check_hostname': True}
@@ -1602,39 +1547,12 @@ def _resolve_hostnames(self):
 
     def _create_thread_pool_executor(self, **kwargs):
         """
-        Create a ThreadPoolExecutor for the cluster. In most cases, the built-in
-        `concurrent.futures.ThreadPoolExecutor` is used.
-
-        Python 3.7+ and Eventlet cause the `concurrent.futures.ThreadPoolExecutor`
-        to hang indefinitely. In that case, the user needs to have the `futurist`
-        package so we can use the `futurist.GreenThreadPoolExecutor` class instead.
+        Create a ThreadPoolExecutor for the cluster.
 
         :param kwargs: All keyword args are passed to the ThreadPoolExecutor constructor.
         :return: A ThreadPoolExecutor instance.
         """
-        tpe_class = ThreadPoolExecutor
-        if sys.version_info[0] >= 3 and sys.version_info[1] >= 7:
-            try:
-                from cassandra.io.eventletreactor import EventletConnection
-                is_eventlet = issubclass(self.connection_class, EventletConnection)
-            except:
-                # Eventlet is not available or can't be detected
-                return tpe_class(**kwargs)
-
-            if is_eventlet:
-                try:
-                    from futurist import GreenThreadPoolExecutor
-                    tpe_class = GreenThreadPoolExecutor
-                except ImportError:
-                    # futurist is not available
-                    raise ImportError(
-                        ("Python 3.7+ and Eventlet cause the `concurrent.futures.ThreadPoolExecutor` "
-                         "to hang indefinitely. If you want to use the Eventlet reactor, you "
-                         "need to install the `futurist` package to allow the driver to use "
-                         "the GreenThreadPoolExecutor. See https://github.com/eventlet/eventlet/issues/508 "
-                         "for more details."))
-
-        return tpe_class(**kwargs)
+        return ThreadPoolExecutor(**kwargs)
 
     def register_user_type(self, keyspace, user_type, klass):
         """

cassandra/connection.py

@@ -35,10 +35,7 @@
 from cassandra.client_routes import _ClientRoutesHandler
 from cassandra.protocol_features import ProtocolFeatures
 
-if 'gevent.monkey' in sys.modules:
-    from gevent.queue import Queue, Empty
-else:
-    from queue import Queue, Empty  # noqa
+from queue import Queue, Empty  # noqa
 
 from cassandra import ConsistencyLevel, AuthenticationFailed, OperationTimedOut, ProtocolVersion
 from cassandra.marshal import int32_pack

cassandra/datastax/cloud/__init__.py

@@ -75,38 +75,44 @@ def from_dict(cls, d):
         return c
 
 
-def get_cloud_config(cloud_config, create_pyopenssl_context=False):
+def get_cloud_config(cloud_config):
     if not _HAS_SSL:
         raise DriverException("A Python installation with SSL is required to connect to a cloud cluster.")
 
     if 'secure_connect_bundle' not in cloud_config:
         raise ValueError("The cloud config doesn't have a secure_connect_bundle specified.")
 
     try:
-        config = read_cloud_config_from_zip(cloud_config, create_pyopenssl_context)
-    except BadZipFile:
-        raise ValueError("Unable to open the zip file for the cloud config. Check your secure connect bundle.")
+        config = read_cloud_config_from_zip(cloud_config)
+    except BadZipFile as err:
+        raise ValueError("Unable to open the zip file for the cloud config. Check your secure connect bundle.") from err
 
     config = read_metadata_info(config, cloud_config)
-    if create_pyopenssl_context:
-        config.ssl_context = config.pyopenssl_context
     return config
 
 
-def read_cloud_config_from_zip(cloud_config, create_pyopenssl_context):
+def _safe_extractall(zipfile, tmp_dir):
+    for member in zipfile.namelist():
+        target = os.path.realpath(os.path.join(tmp_dir, member))
+        if not target.startswith(os.path.realpath(tmp_dir) + os.sep):
+            raise ValueError("Secure connect bundle contains an unsafe path: %s" % member)
+    zipfile.extractall(path=tmp_dir)
+
+
+def read_cloud_config_from_zip(cloud_config):
     secure_bundle = cloud_config['secure_connect_bundle']
     use_default_tempdir = cloud_config.get('use_default_tempdir', None)
     with ZipFile(secure_bundle) as zipfile:
         base_dir = tempfile.gettempdir() if use_default_tempdir else os.path.dirname(secure_bundle)
         tmp_dir = tempfile.mkdtemp(dir=base_dir)
         try:
-            zipfile.extractall(path=tmp_dir)
-            return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config, create_pyopenssl_context)
+            _safe_extractall(zipfile, tmp_dir)
+            return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config)
         finally:
             shutil.rmtree(tmp_dir)
 
 
-def parse_cloud_config(path, cloud_config, create_pyopenssl_context):
+def parse_cloud_config(path, cloud_config):
     with open(path, 'r') as stream:
         data = json.load(stream)
 
@@ -120,11 +126,7 @@ def parse_cloud_config(path, cloud_config, create_pyopenssl_context):
         ca_cert_location = os.path.join(config_dir, 'ca.crt')
         cert_location = os.path.join(config_dir, 'cert')
         key_location = os.path.join(config_dir, 'key')
-        # Regardless of if we create a pyopenssl context, we still need the builtin one
-        # to connect to the metadata service
         config.ssl_context = _ssl_context_from_cert(ca_cert_location, cert_location, key_location)
-        if create_pyopenssl_context:
-            config.pyopenssl_context = _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location)
 
     return config
 
@@ -175,18 +177,3 @@ def _ssl_context_from_cert(ca_cert_location, cert_location, key_location):
 
     return ssl_context
 
-
-def _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location):
-    try:
-        from OpenSSL import SSL
-    except ImportError as e:
-        raise ImportError(
-            "PyOpenSSL must be installed to connect to Astra with the Eventlet or Twisted event loops")\
-            .with_traceback(e.__traceback__)
-    ssl_context = SSL.Context(SSL.TLSv1_METHOD)
-    ssl_context.set_verify(SSL.VERIFY_PEER, callback=lambda _1, _2, _3, _4, ok: ok)
-    ssl_context.use_certificate_file(cert_location)
-    ssl_context.use_privatekey_file(key_location)
-    ssl_context.load_verify_locations(ca_cert_location)
-
-    return ssl_context

cassandra/datastax/insights/reporter.py

@@ -142,11 +142,7 @@ def _get_startup_data(self):
         cert_validation = None
         try:
             if self._session.cluster.ssl_context:
-                if isinstance(self._session.cluster.ssl_context, ssl.SSLContext):
-                    cert_validation = self._session.cluster.ssl_context.verify_mode == ssl.CERT_REQUIRED
-                else:  # pyopenssl
-                    from OpenSSL import SSL
-                    cert_validation = self._session.cluster.ssl_context.get_verify_mode() != SSL.VERIFY_NONE
+                cert_validation = self._session.cluster.ssl_context.verify_mode == ssl.CERT_REQUIRED
             elif self._session.cluster.ssl_options:
                 cert_validation = self._session.cluster.ssl_options.get('cert_reqs') == ssl.CERT_REQUIRED
         except Exception as e: