GG-32660: Port review changes from several donations (#13)

Co-authored-by: Ivan Dashchinskiy <ivandasch@gmail.com>
Co-authored-by: Pavel Tupitsyn <ptupitsyn@apache.org>
Co-authored-by: Ilya Kasnacheev <ilya.kasnacheev@gmail.com>

Author: 4 people

pygridgain/api/key_value.py

@@ -977,10 +977,7 @@ def cache_get_size(
      otherwise.
     """
     if not isinstance(peek_modes, (list, tuple)):
-        if peek_modes == 0:
-            peek_modes = []
-        else:
-            peek_modes = [peek_modes]
+        peek_modes = [peek_modes] if peek_modes else []
 
     query_struct = Query(
         OP_CACHE_GET_SIZE,
@@ -1035,10 +1032,7 @@ def cache_local_peek(
      (null if not found).
     """
     if not isinstance(peek_modes, (list, tuple)):
-        if peek_modes == 0:
-            peek_modes = []
-        else:
-            peek_modes = [peek_modes]
+        peek_modes = [peek_modes] if peek_modes else []
 
     query_struct = Query(
         OP_CACHE_LOCAL_PEEK,

pygridgain/api/sql.py

@@ -416,10 +416,7 @@ def sql_fields_cursor_get_page(
             'cursor': cursor,
         },
         response_config=[
-            ('data', StructArray([
-                ('field_{}'.format(i), AnyDataObject)
-                for i in range(field_count)
-            ])),
+            ('data', StructArray([(f'field_{i}', AnyDataObject) for i in range(field_count)])),
             ('more', Bool),
         ]
     )

pygridgain/binary.py

@@ -69,12 +69,10 @@ def schema_id(self) -> int:
     def __new__(cls, *args, **kwargs) -> Any:
         # allow all items in Binary Object schema to be populated as optional
         # arguments to `__init__()` with sensible defaults.
-        attributes = {
-            k: attr.ib(
-                type=getattr(v, 'pythonic', type(None)),
-                default=getattr(v, 'default', None),
-            ) for k, v in cls.schema.items()
-        }
+        attributes = {}
+        for k, v in cls.schema.items():
+            attributes[k] = attr.ib(type=getattr(v, 'pythonic', type(None)), default=getattr(v, 'default', None))
+
         attributes.update({'version': attr.ib(type=int, default=1)})
         cls = attr.s(cls, these=attributes)
         # skip parameters
@@ -151,10 +149,7 @@ def _build(self, client: 'Client' = None) -> int:
                 header.flags |= BinaryObject.OFFSET_ONE_BYTE
             elif max(offsets) < 65535:
                 header.flags |= BinaryObject.OFFSET_TWO_BYTES
-            schema_class = (
-                BinaryObject.schema_type(header.flags)
-                * len(offsets)
-            )
+            schema_class = BinaryObject.schema_type(header.flags) * len(offsets)
             schema = schema_class()
             if compact_footer:
                 for i, offset in enumerate(offsets):

pygridgain/cache.py

@@ -25,7 +25,7 @@
     connection_errors,
 )
 from .utils import (
-    cache_id, get_field_by_id, is_wrapped, select_version,
+    cache_id, get_field_by_id, is_wrapped,
     status_to_exception, unsigned, unwrap_binary,
 )
 from .api.cache_config import (
@@ -234,7 +234,6 @@ def _get_affinity(self, conn: 'Connection') -> Dict:
 
         return result
 
-    @select_version
     def get_best_node(
         self, key: Any = None, key_hint: 'GridGainDataType' = None,
     ) -> 'Connection':
@@ -257,8 +256,6 @@ def get_best_node(
             if key_hint is None:
                 key_hint = AnyDataObject.map_python_type(key)
 
-            parts = -1
-
             if self.affinity['version'] < self._client.affinity_version:
                 # update partition mapping
                 while True:
@@ -280,9 +277,10 @@ def get_best_node(
                 del self.affinity['partition_mapping']
 
                 # calculate the number of partitions
-                parts = sum(
-                    [len(p) for _, p in self.affinity['node_mapping'].items()]
-                ) if 'node_mapping' in self.affinity else 0
+                parts = 0
+                if 'node_mapping' in self.affinity:
+                    for p in self.affinity['node_mapping'].values():
+                        parts += len(p)
 
                 self.affinity['number_of_partitions'] = parts
             else:
@@ -314,26 +312,24 @@ def get_best_node(
 
             # search for connection
             try:
-                node_uuid = next(
-                    u for u, p
-                    in self.affinity['node_mapping'].items()
-                    if part in p
-                )
-                best_conn = next(
-                    n for n in conn.client._nodes if n.uuid == node_uuid
-                )
-                if best_conn.alive:
-                    conn = best_conn
-            except (StopIteration, KeyError):
+                node_uuid, best_conn = None, None
+                for u, p in self.affinity['node_mapping'].items():
+                    if part in p:
+                        node_uuid = u
+                        break
+
+                if node_uuid:
+                    for n in conn.client._nodes:
+                        if n.uuid == node_uuid:
+                            best_conn = n
+                            break
+                    if best_conn and best_conn.alive:
+                        conn = best_conn
+            except KeyError:
                 pass
 
         return conn
 
-    def get_best_node_130(self, *args, **kwargs):
-        return self.client.random_node
-
-    get_best_node_120 = get_best_node_130
-
     @status_to_exception(CacheError)
     def get(self, key, key_hint: object = None) -> Any:
         """
@@ -404,9 +400,7 @@ def put_all(self, pairs: dict):
          to save. Each key or value can be an item of representable
          Python type or a tuple of (item, hint),
         """
-        return cache_put_all(
-            self.get_best_node(), self._cache_id, pairs
-        )
+        return cache_put_all(self.get_best_node(), self._cache_id, pairs)
 
     @status_to_exception(CacheError)
     def replace(

pygridgain/client.py

@@ -43,6 +43,7 @@
 from collections import defaultdict, OrderedDict
 import random
 import re
+from itertools import chain
 from typing import Dict, Iterable, List, Optional, Tuple, Type, Union
 
 from .api.binary import get_binary_type, put_binary_type
@@ -57,7 +58,7 @@
     BinaryTypeError, CacheError, ReconnectError, SQLError, connection_errors,
 )
 from .utils import (
-    capitalize, entity_id, schema_id, process_delimiter, select_version,
+    capitalize, entity_id, schema_id, process_delimiter,
     status_to_exception, is_iterable,
 )
 from .binary import GenericObjectMeta
@@ -131,7 +132,14 @@ def get_protocol_version(self) -> Optional[Tuple]:
 
     @property
     def partition_aware(self):
-        return self._partition_aware
+        return self._partition_aware and self.partition_awareness_supported_by_protocol
+
+    @property
+    def partition_awareness_supported_by_protocol(self):
+        # TODO: Need to re-factor this. I believe, we need separate class or
+        #  set of functions to work with protocol versions without manually
+        #  comparing versions with just some random tuples
+        return self.protocol_version is not None and self.protocol_version >= (1, 4, 0)
 
     def connect(self, *args):
         """
@@ -158,23 +166,20 @@ def connect(self, *args):
         # the following code is quite twisted, because the protocol version
         # is initially unknown
 
-        # TODO: open first node in foregroung, others − in background
+        # TODO: open first node in foreground, others − in background
         for i, node in enumerate(nodes):
             host, port = node
             conn = Connection(self, **self._connection_args)
             conn.host = host
             conn.port = port
 
             try:
-                if (
-                    self.protocol_version is None
-                    or self.protocol_version >= (1, 4, 0)
-                ):
+                if self.protocol_version is None or self.partition_aware:
                     # open connection before adding to the pool
                     conn.connect(host, port)
 
                     # now we have the protocol version
-                    if self.protocol_version < (1, 4, 0):
+                    if not self.partition_aware:
                         # do not try to open more nodes
                         self._current_node = i
                     else:
@@ -186,10 +191,7 @@ def connect(self, *args):
 
             except connection_errors:
                 conn._fail()
-                if (
-                    self.protocol_version
-                    and self.protocol_version >= (1, 4, 0)
-                ):
+                if self.partition_aware:
                     # schedule the reconnection
                     conn.reconnect()
 
@@ -204,7 +206,6 @@ def close(self):
         self._nodes.clear()
 
     @property
-    @select_version
     def random_node(self) -> Connection:
         """
         Returns random usable node.
@@ -213,46 +214,44 @@ def random_node(self) -> Connection:
         extend the `pygridgain` capabilities (with additional testing,
         logging, examining connections, et c.) you probably should not use it.
         """
-        try:
-            return random.choice(
-                list(n for n in self._nodes if n.alive)
-            )
-        except IndexError:
-            # cannot choose from an empty sequence
-            raise ReconnectError('Can not reconnect: out of nodes.') from None
-
-    def random_node_130(self):
-        # it actually returns the next usable node, but the name stands for
-        # the code unification reason
-        node = self._nodes[self._current_node]
-        if node.alive:
-            return node
-
-        # close current (supposedly failed) node
-        self._nodes[self._current_node].close()
-
-        # advance the node index
-        self._current_node += 1
-        if self._current_node >= len(self._nodes):
-            self._current_node = 0
-
-        # prepare the list of node indexes to try to connect to
-        seq = list(range(len(self._nodes)))
-        seq = seq[self._current_node:] + seq[:self._current_node]
-
-        for i in seq:
-            node = self._nodes[i]
+        if self.partition_aware:
+            # if partition awareness is used just pick a random connected node
             try:
-                node.connect(node.host, node.port)
-            except connection_errors:
-                pass
-            else:
+                return random.choice(
+                    list(n for n in self._nodes if n.alive)
+                )
+            except IndexError:
+                # cannot choose from an empty sequence
+                raise ReconnectError('Can not reconnect: out of nodes.') from None
+        else:
+            # if partition awareness is not used then just return the current
+            # node if it's alive or the next usable node if connection with the
+            # current is broken
+            node = self._nodes[self._current_node]
+            if node.alive:
                 return node
 
-        # no nodes left
-        raise ReconnectError('Can not reconnect: out of nodes.')
-
-    random_node_120 = random_node_130
+            # close current (supposedly failed) node
+            self._nodes[self._current_node].close()
+
+            # advance the node index
+            self._current_node += 1
+            if self._current_node >= len(self._nodes):
+                self._current_node = 0
+
+            # prepare the list of node indexes to try to connect to
+            num_nodes = len(self._nodes)
+            for i in chain(range(self._current_node, num_nodes), range(self._current_node)):
+                node = self._nodes[i]
+                try:
+                    node.connect(node.host, node.port)
+                except connection_errors:
+                    pass
+                else:
+                    return node
+
+            # no nodes left
+            raise ReconnectError('Can not reconnect: out of nodes.')
 
     @status_to_exception(BinaryTypeError)
     def get_binary_type(self, binary_type: Union[str, int]) -> dict: