Merge pull request #678 from a-ungurianu/devx/bye_bye_six

refactor: Remove vestiges of python2 support

Author: a-ungurianu

kazoo/client.py

@@ -7,8 +7,6 @@
 import re
 import warnings
 
-import six
-
 from kazoo.exceptions import (
     AuthFailedError,
     ConfigurationError,
@@ -66,9 +64,6 @@
 from kazoo.recipe.watchers import ChildrenWatch, DataWatch
 
 
-string_types = six.string_types
-bytes_types = (six.binary_type,)
-
 CLOSED_STATES = (
     KeeperState.EXPIRED_SESSION,
     KeeperState.AUTH_FAILED,
@@ -415,10 +410,10 @@ def _reset(self):
 
     def _reset_watchers(self):
         watchers = []
-        for child_watchers in six.itervalues(self._child_watchers):
+        for child_watchers in self._child_watchers.values():
             watchers.extend(child_watchers)
 
-        for data_watchers in six.itervalues(self._data_watchers):
+        for data_watchers in self._data_watchers.values():
             watchers.extend(data_watchers)
 
         self._child_watchers = defaultdict(set)
@@ -821,7 +816,7 @@ def _is_valid(version):
         version = _try_fetch()
         if _is_valid(version):
             return version
-        for _i in six.moves.range(0, retries):
+        for _i in range(0, retries):
             version = _try_fetch()
             if _is_valid(version):
                 return version
@@ -854,9 +849,9 @@ def add_auth_async(self, scheme, credential):
         :rtype: :class:`~kazoo.interfaces.IAsyncResult`
 
         """
-        if not isinstance(scheme, string_types):
+        if not isinstance(scheme, str):
             raise TypeError("Invalid type for 'scheme' (string expected)")
-        if not isinstance(credential, string_types):
+        if not isinstance(credential, str):
             raise TypeError("Invalid type for 'credential' (string expected)")
 
         # we need this auth data to re-authenticate on reconnect
@@ -1034,15 +1029,15 @@ def create_async(
         if acl is None and self.default_acl:
             acl = self.default_acl
 
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
         if acl and (
             isinstance(acl, ACL) or not isinstance(acl, (tuple, list))
         ):
             raise TypeError(
                 "Invalid type for 'acl' (acl must be a tuple/list" " of ACL's"
             )
-        if value is not None and not isinstance(value, bytes_types):
+        if value is not None and not isinstance(value, bytes):
             raise TypeError("Invalid type for 'value' (must be a byte string)")
         if not isinstance(ephemeral, bool):
             raise TypeError("Invalid type for 'ephemeral' (bool expected)")
@@ -1205,7 +1200,7 @@ def exists_async(self, path, watch=None):
         :rtype: :class:`~kazoo.interfaces.IAsyncResult`
 
         """
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
         if watch and not callable(watch):
             raise TypeError("Invalid type for 'watch' (must be a callable)")
@@ -1248,7 +1243,7 @@ def get_async(self, path, watch=None):
         :rtype: :class:`~kazoo.interfaces.IAsyncResult`
 
         """
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
         if watch and not callable(watch):
             raise TypeError("Invalid type for 'watch' (must be a callable)")
@@ -1304,7 +1299,7 @@ def get_children_async(self, path, watch=None, include_data=False):
         :rtype: :class:`~kazoo.interfaces.IAsyncResult`
 
         """
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
         if watch and not callable(watch):
             raise TypeError("Invalid type for 'watch' (must be a callable)")
@@ -1346,7 +1341,7 @@ def get_acls_async(self, path):
         :rtype: :class:`~kazoo.interfaces.IAsyncResult`
 
         """
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
 
         async_result = self.handler.async_result()
@@ -1389,7 +1384,7 @@ def set_acls_async(self, path, acls, version=-1):
         :rtype: :class:`~kazoo.interfaces.IAsyncResult`
 
         """
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
         if isinstance(acls, ACL) or not isinstance(acls, (tuple, list)):
             raise TypeError(
@@ -1447,9 +1442,9 @@ def set_async(self, path, value, version=-1):
         :rtype: :class:`~kazoo.interfaces.IAsyncResult`
 
         """
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
-        if value is not None and not isinstance(value, bytes_types):
+        if value is not None and not isinstance(value, bytes):
             raise TypeError("Invalid type for 'value' (must be a byte string)")
         if not isinstance(version, int):
             raise TypeError("Invalid type for 'version' (int expected)")
@@ -1523,7 +1518,7 @@ def delete_async(self, path, version=-1):
         :rtype: :class:`~kazoo.interfaces.IAsyncResult`
 
         """
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
         if not isinstance(version, int):
             raise TypeError("Invalid type for 'version' (int expected)")
@@ -1632,11 +1627,11 @@ def reconfig_async(self, joining, leaving, new_members, from_config):
         :rtype: :class:`~kazoo.interfaces.IAsyncResult`
 
         """
-        if joining and not isinstance(joining, string_types):
+        if joining and not isinstance(joining, str):
             raise TypeError("Invalid type for 'joining' (string expected)")
-        if leaving and not isinstance(leaving, string_types):
+        if leaving and not isinstance(leaving, str):
             raise TypeError("Invalid type for 'leaving' (string expected)")
-        if new_members and not isinstance(new_members, string_types):
+        if new_members and not isinstance(new_members, str):
             raise TypeError(
                 "Invalid type for 'new_members' (string " "expected)"
             )
@@ -1690,13 +1685,13 @@ def create(
         if acl is None and self.client.default_acl:
             acl = self.client.default_acl
 
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
         if acl and not isinstance(acl, (tuple, list)):
             raise TypeError(
                 "Invalid type for 'acl' (acl must be a tuple/list" " of ACL's"
             )
-        if not isinstance(value, bytes_types):
+        if not isinstance(value, bytes):
             raise TypeError("Invalid type for 'value' (must be a byte string)")
         if not isinstance(ephemeral, bool):
             raise TypeError("Invalid type for 'ephemeral' (bool expected)")
@@ -1722,7 +1717,7 @@ def delete(self, path, version=-1):
         `recursive`.
 
         """
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
         if not isinstance(version, int):
             raise TypeError("Invalid type for 'version' (int expected)")
@@ -1733,9 +1728,9 @@ def set_data(self, path, value, version=-1):
         arguments as :meth:`KazooClient.set`.
 
         """
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
-        if not isinstance(value, bytes_types):
+        if not isinstance(value, bytes):
             raise TypeError("Invalid type for 'value' (must be a byte string)")
         if not isinstance(version, int):
             raise TypeError("Invalid type for 'version' (int expected)")
@@ -1750,7 +1745,7 @@ def check(self, path, version):
         does not match the specified version.
 
         """
-        if not isinstance(path, string_types):
+        if not isinstance(path, str):
             raise TypeError("Invalid type for 'path' (string expected)")
         if not isinstance(version, int):
             raise TypeError("Invalid type for 'version' (int expected)")

kazoo/handlers/eventlet.py

@@ -1,6 +1,7 @@
 """A eventlet based handler."""
 from __future__ import absolute_import
 
+import atexit
 import contextlib
 import logging
 
@@ -12,7 +13,6 @@
 from eventlet import queue as green_queue
 
 from kazoo.handlers import utils
-import kazoo.python2atexit as python2atexit
 from kazoo.handlers.utils import selector_select
 
 LOG = logging.getLogger(__name__)
@@ -140,15 +140,15 @@ def start(self):
             w = eventlet.spawn(self._process_callback_queue)
             self._workers.append((w, self.callback_queue))
             self._started = True
-            python2atexit.register(self.stop)
+            atexit.register(self.stop)
 
     def stop(self):
         while self._workers:
             w, q = self._workers.pop()
             q.put(_STOP)
             w.wait()
         self._started = False
-        python2atexit.unregister(self.stop)
+        atexit.unregister(self.stop)
 
     def socket(self, *args, **kwargs):
         return utils.create_tcp_socket(green_socket)

kazoo/handlers/gevent.py

@@ -1,6 +1,7 @@
 """A gevent based handler."""
 from __future__ import absolute_import
 
+import atexit
 import logging
 
 import gevent
@@ -13,13 +14,9 @@
 
 from kazoo.handlers.utils import selector_select
 
-try:
-    from gevent.lock import Semaphore, RLock
-except ImportError:
-    from gevent.coros import Semaphore, RLock
+from gevent.lock import Semaphore, RLock
 
 from kazoo.handlers import utils
-from kazoo import python2atexit
 
 _using_libevent = gevent.__version__.startswith("0.")
 
@@ -104,7 +101,7 @@ def start(self):
             for queue in (self.callback_queue,):
                 w = self._create_greenlet_worker(queue)
                 self._workers.append(w)
-            python2atexit.register(self.stop)
+            atexit.register(self.stop)
 
     def stop(self):
         """Stop the greenlet workers and empty all queues."""
@@ -124,7 +121,7 @@ def stop(self):
             # Clear the queues
             self.callback_queue = self.queue_impl()  # pragma: nocover
 
-            python2atexit.unregister(self.stop)
+            atexit.unregister(self.stop)
 
     def select(self, *args, **kwargs):
         return selector_select(

kazoo/handlers/threading.py

@@ -12,21 +12,16 @@
 """
 from __future__ import absolute_import
 
+import atexit
 import logging
+import queue
 import socket
 import threading
 import time
 
-import six
-
-import kazoo.python2atexit as python2atexit
 from kazoo.handlers import utils
 from kazoo.handlers.utils import selector_select
 
-try:
-    import Queue
-except ImportError:  # pragma: nocover
-    import queue as Queue
 
 # sentinel objects
 _STOP = object()
@@ -35,11 +30,11 @@
 
 
 def _to_fileno(obj):
-    if isinstance(obj, six.integer_types):
+    if isinstance(obj, int):
         fd = int(obj)
     elif hasattr(obj, "fileno"):
         fd = obj.fileno()
-        if not isinstance(fd, six.integer_types):
+        if not isinstance(fd, int):
             raise TypeError("fileno() returned a non-integer")
         fd = int(fd)
     else:
@@ -98,8 +93,8 @@ class SequentialThreadingHandler(object):
     name = "sequential_threading_handler"
     timeout_exception = KazooTimeoutError
     sleep_func = staticmethod(time.sleep)
-    queue_impl = Queue.Queue
-    queue_empty = Queue.Empty
+    queue_impl = queue.Queue
+    queue_empty = queue.Empty
 
     def __init__(self):
         """Create a :class:`SequentialThreadingHandler` instance"""
@@ -113,19 +108,19 @@ def __init__(self):
     def running(self):
         return self._running
 
-    def _create_thread_worker(self, queue):
+    def _create_thread_worker(self, work_queue):
         def _thread_worker():  # pragma: nocover
             while True:
                 try:
-                    func = queue.get()
+                    func = work_queue.get()
                     try:
                         if func is _STOP:
                             break
                         func()
                     except Exception:
                         log.exception("Exception in worker queue thread")
                     finally:
-                        queue.task_done()
+                        work_queue.task_done()
                         del func  # release before possible idle
                 except self.queue_empty:
                     continue
@@ -142,11 +137,11 @@ def start(self):
             # Spawn our worker threads, we have
             # - A callback worker for watch events to be called
             # - A completion worker for completion events to be called
-            for queue in (self.completion_queue, self.callback_queue):
-                w = self._create_thread_worker(queue)
+            for work_queue in (self.completion_queue, self.callback_queue):
+                w = self._create_thread_worker(work_queue)
                 self._workers.append(w)
             self._running = True
-            python2atexit.register(self.stop)
+            atexit.register(self.stop)
 
     def stop(self):
         """Stop the worker threads and empty all queues."""
@@ -156,8 +151,8 @@ def stop(self):
 
             self._running = False
 
-            for queue in (self.completion_queue, self.callback_queue):
-                queue.put(_STOP)
+            for work_queue in (self.completion_queue, self.callback_queue):
+                work_queue.put(_STOP)
 
             self._workers.reverse()
             while self._workers:
@@ -167,7 +162,7 @@ def stop(self):
             # Clear the queues
             self.callback_queue = self.queue_impl()
             self.completion_queue = self.queue_impl()
-            python2atexit.unregister(self.stop)
+            atexit.unregister(self.stop)
 
     def select(self, *args, **kwargs):
         return selector_select(*args, **kwargs)