def _translate_failures():
    """Translates common redis exceptions into tooz exceptions."""
    try:
        yield
    except (exceptions.ConnectionError, exceptions.TimeoutError) as e:
        raise coordination.ToozConnectionError(utils.exception_message(e))
    except exceptions.RedisError as e:
        raise coordination.ToozError(utils.exception_message(e))

 def _update_capabilities_handler(async_result, timeout,
                                  timeout_exception, group_id, member_id):
     try:
         async_result.get(block=True, timeout=timeout)
     except timeout_exception as e:
         raise coordination.OperationTimedOut(utils.exception_message(e))
     except exceptions.NoNodeError:
         raise coordination.MemberNotJoined(group_id, member_id)
     except exceptions.ZookeeperError as e:
         raise coordination.ToozError(utils.exception_message(e))

 def _get_groups_handler(async_result, timeout, timeout_exception):
     try:
         group_ids = async_result.get(block=True, timeout=timeout)
     except timeout_exception as e:
         raise coordination.OperationTimedOut(utils.exception_message(e))
     except exceptions.NoNodeError:
         raise coordination.ToozError("tooz namespace has not been created")
     except exceptions.ZookeeperError as e:
         raise coordination.ToozError(utils.exception_message(e))
     else:
         return set(g.encode('ascii') for g in group_ids)

 def _delete_group_handler(async_result, timeout,
                           timeout_exception, group_id):
     try:
         async_result.get(block=True, timeout=timeout)
     except timeout_exception as e:
         raise coordination.OperationTimedOut(utils.exception_message(e))
     except exceptions.NoNodeError:
         raise coordination.GroupNotCreated(group_id)
     except exceptions.NotEmptyError:
         raise coordination.GroupNotEmpty(group_id)
     except exceptions.ZookeeperError as e:
         raise coordination.ToozError(utils.exception_message(e))

 def _join_group_handler(async_result, timeout,
                         timeout_exception, group_id, member_id):
     try:
         async_result.get(block=True, timeout=timeout)
     except timeout_exception as e:
         raise coordination.OperationTimedOut(utils.exception_message(e))
     except exceptions.NodeExistsError:
         raise coordination.MemberAlreadyExist(group_id, member_id)
     except exceptions.NoNodeError:
         raise coordination.GroupNotCreated(group_id)
     except exceptions.ZookeeperError as e:
         raise coordination.ToozError(utils.exception_message(e))

 def _create_group_handler(async_result, timeout,
                           timeout_exception, group_id):
     try:
         async_result.get(block=True, timeout=timeout)
     except timeout_exception as e:
         raise coordination.OperationTimedOut(utils.exception_message(e))
     except exceptions.NodeExistsError:
         raise coordination.GroupAlreadyExist(group_id)
     except exceptions.NoNodeError:
         raise coordination.ToozError("tooz namespace has not been created")
     except exceptions.ZookeeperError as e:
         raise coordination.ToozError(utils.exception_message(e))

 def _get_members_handler(async_result, timeout,
                          timeout_exception, group_id):
     try:
         members_ids = async_result.get(block=True, timeout=timeout)
     except timeout_exception as e:
         raise coordination.OperationTimedOut(utils.exception_message(e))
     except exceptions.NoNodeError:
         raise coordination.GroupNotCreated(group_id)
     except exceptions.ZookeeperError as e:
         raise coordination.ToozError(utils.exception_message(e))
     else:
         return set(m.encode('ascii') for m in members_ids)

 def _get_member_capabilities_handler(cls, async_result, timeout,
                                      timeout_exception, group_id,
                                      member_id):
     try:
         capabilities = async_result.get(block=True, timeout=timeout)[0]
     except timeout_exception as e:
         raise coordination.OperationTimedOut(utils.exception_message(e))
     except exceptions.NoNodeError:
         raise coordination.MemberNotJoined(group_id, member_id)
     except exceptions.ZookeeperError as e:
         raise coordination.ToozError(utils.exception_message(e))
     else:
         return cls._loads(capabilities)

 def _leave_group_handler(async_result, timeout,
                          timeout_exception, group_id, member_id):
     try:
         async_result.get(block=True, timeout=timeout)
     except timeout_exception as e:
         coordination.raise_with_cause(coordination.OperationTimedOut,
                                       utils.exception_message(e),
                                       cause=e)
     except exceptions.NoNodeError:
         raise coordination.MemberNotJoined(group_id, member_id)
     except exceptions.ZookeeperError as e:
         coordination.raise_with_cause(coordination.ToozError,
                                       utils.exception_message(e),
                                       cause=e)

    def get_connection(parsed_url, options):
        host = parsed_url.hostname
        port = parsed_url.port
        dbname = parsed_url.path[1:]
        username = parsed_url.username
        password = parsed_url.password
        unix_socket = options.get("unix_socket")

        try:
            if unix_socket:
                return pymysql.Connect(unix_socket=unix_socket,
                                       port=port,
                                       user=username,
                                       passwd=password,
                                       database=dbname)
            else:
                return pymysql.Connect(host=host,
                                       port=port,
                                       user=username,
                                       passwd=password,
                                       database=dbname)
        except (pymysql.err.OperationalError, pymysql.err.InternalError) as e:
            coordination.raise_with_cause(coordination.ToozConnectionError,
                                          utils.exception_message(e),
                                          cause=e)

def _translate_failures():
    try:
        yield
    except EnvironmentError as e:
        coordination.raise_with_cause(coordination.ToozError,
                                      utils.exception_message(e),
                                      cause=e)

 def get(self, timeout=10):
     try:
         return self._fut.result(timeout=timeout)
     except futures.TimeoutError as e:
         coordination.raise_with_cause(coordination.OperationTimedOut,
                                       utils.exception_message(e),
                                       cause=e)

        def _lock():
            # NOTE(sileht): mysql-server (<5.7.5) allows only one lock per
            # connection at a time:
            #  select GET_LOCK("a", 0);
            #  select GET_LOCK("b", 0); <-- this release lock "a" ...
            # Or
            #  select GET_LOCK("a", 0);
            #  select GET_LOCK("a", 0); release and lock again "a"
            #
            # So, we track locally the lock status with self.acquired
            if self.acquired is True:
                if blocking:
                    raise _retry.Retry
                return False

            try:
                with self._conn as cur:
                    cur.execute("SELECT GET_LOCK(%s, 0);", self.name)
                    # Can return NULL on error
                    if cur.fetchone()[0] is 1:
                        self.acquired = True
                        return True
            except pymysql.MySQLError as e:
                coordination.raise_with_cause(coordination.ToozError, utils.exception_message(e), cause=e)

            if blocking:
                raise _retry.Retry
            return False

 def acquire(self, blocking=True):
     if blocking is False:
         try:
             cur = self._conn.cursor()
             cur.execute("SELECT GET_LOCK(%s, 0);", self.name)
             # Can return NULL on error
             if cur.fetchone()[0] is 1:
                 return True
             return False
         except pymysql.MySQLError as e:
             raise coordination.ToozError(utils.exception_message(e))
     else:
         def _acquire():
             try:
                 cur = self._conn.cursor()
                 cur.execute("SELECT GET_LOCK(%s, 0);", self.name)
                 if cur.fetchone()[0] is 1:
                     return True
             except pymysql.MySQLError as e:
                 raise coordination.ToozError(utils.exception_message(e))
             raise _retry.Retry
         kwargs = _retry.RETRYING_KWARGS.copy()
         if blocking is not True:
             kwargs['stop_max_delay'] = blocking
         return _retry.Retrying(**kwargs).call(_acquire)

 def release(self):
     try:
         with self._conn as cur:
             cur.execute("SELECT RELEASE_LOCK(%s);", self.name)
             return cur.fetchone()[0]
     except pymysql.MySQLError as e:
         raise coordination.ToozError(utils.exception_message(e))

 def _start(self):
     self._executor = futures.ThreadPoolExecutor(max_workers=1)
     try:
         self._client = self._make_client(self._parsed_url, self._options,
                                          self.timeout)
     except exceptions.RedisError as e:
         raise coordination.ToozConnectionError(utils.exception_message(e))
     else:
         # Ensure that the server is alive and not dead, this does not
         # ensure the server will always be alive, but does insure that it
         # at least is alive once...
         with _translate_failures():
             self._server_info = self._client.info()
         # Validate we have a good enough redis version we are connected
         # to so that the basic set of features we support will actually
         # work (instead of blowing up).
         new_enough, redis_version = self._check_fetch_redis_version(
             self._MIN_VERSION)
         if not new_enough:
             raise tooz.NotImplemented("Redis version greater than or"
                                       " equal to '%s' is required"
                                       " to use this driver; '%s' is"
                                       " being used which is not new"
                                       " enough" % (self._MIN_VERSION,
                                                    redis_version))
         self.heartbeat()
         self._started = True

 def _safe_stop(coord):
     try:
         coord.stop()
     except coordination.ToozError as e:
         message = utils.exception_message(e)
         if (message != 'Can not stop a driver which has not'
                        ' been started'):
             raise

 def _acquire():
     try:
         cur = self._conn.cursor()
         cur.execute("SELECT GET_LOCK(%s, 0);", self.name)
         if cur.fetchone()[0] is 1:
             return True
     except pymysql.MySQLError as e:
         raise coordination.ToozError(utils.exception_message(e))
     raise _retry.Retry

 def release(self):
     try:
         with self._conn as cur:
             cur.execute("SELECT RELEASE_LOCK(%s);", self.name)
             cur.fetchone()
             self.acquired = False
     except pymysql.MySQLError as e:
         coordination.raise_with_cause(coordination.ToozError,
                                       utils.exception_message(e),
                                       cause=e)

 def get(self, timeout=10):
     try:
         # Late translate the common failures since the redis client
         # may throw things that we can not catch in the callbacks where
         # it is used (especially one that uses the transaction
         # method).
         with _translate_failures():
             return self._fut.result(timeout=timeout)
     except futures.TimeoutError as e:
         raise coordination.OperationTimedOut(utils.exception_message(e))