IGNITE-12467 Implement transactions, rewrite async connections using protocol and transport - Fixes #40.

Author: ivandasch

docs/source/pyignite.datatypes.transactions.rst

@@ -0,0 +1,21 @@
+..  Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You under the Apache License, Version 2.0
+    (the "License"); you may not use this file except in compliance with
+    the License.  You may obtain a copy of the License at
+
+..      http://www.apache.org/licenses/LICENSE-2.0
+
+..  Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    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.
+
+pyignite.datatypes.transactions module
+=======================================
+
+.. automodule:: pyignite.datatypes.transactions
+    :members:
+    :show-inheritance:

docs/source/pyignite.rst

@@ -41,6 +41,7 @@ Submodules
     pyignite.aio_client
     pyignite.cluster
     pyignite.aio_cluster
+    pyignite.transaction
     pyignite.cursors
     pyignite.exceptions
 

docs/source/pyignite.transaction.rst

@@ -0,0 +1,22 @@
+..  Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You under the Apache License, Version 2.0
+    (the "License"); you may not use this file except in compliance with
+    the License.  You may obtain a copy of the License at
+
+..      http://www.apache.org/licenses/LICENSE-2.0
+
+..  Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    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.
+
+pyignite.transaction module
+=========================
+
+.. automodule:: pyignite.transaction
+    :members:
+    :undoc-members:
+    :show-inheritance:

pyignite/aio_cache.py

@@ -15,6 +15,7 @@
 import asyncio
 from typing import Any, Iterable, Optional, Union
 
+from .api.tx_api import get_tx_connection
 from .datatypes import ExpiryPolicy
 from .datatypes.internal import AnyDataObject
 from .exceptions import CacheCreationError, CacheError, ParameterError
@@ -91,6 +92,9 @@ def __init__(self, client: 'AioClient', name: str, expiry_policy: ExpiryPolicy =
         super().__init__(client, name, expiry_policy)
 
     async def _get_best_node(self, key=None, key_hint=None):
+        tx_conn = get_tx_connection()
+        if tx_conn:
+            return tx_conn
         return await self.client.get_best_node(self, key, key_hint)
 
     async def settings(self) -> Optional[dict]:

pyignite/aio_client.py

@@ -14,8 +14,9 @@
 # limitations under the License.
 import asyncio
 import random
+import sys
 from itertools import chain
-from typing import Iterable, Type, Union, Any, Dict
+from typing import Iterable, Type, Union, Any, Dict, Optional
 
 from .aio_cluster import AioCluster
 from .api import cache_get_node_partitions_async
@@ -27,10 +28,11 @@
 from .aio_cache import AioCache, get_cache, create_cache, get_or_create_cache
 from .connection import AioConnection
 from .constants import AFFINITY_RETRIES, AFFINITY_DELAY
-from .datatypes import BinaryObject
-from .exceptions import BinaryTypeError, CacheError, ReconnectError, connection_errors
-from .queries.query import CacheInfo
+from .datatypes import BinaryObject, TransactionConcurrency, TransactionIsolation
+from .exceptions import BinaryTypeError, CacheError, ReconnectError, connection_errors, NotSupportedError
+from .queries.cache_info import CacheInfo
 from .stream import AioBinaryStream, READ_BACKWARD
+from .transaction import AioTransaction
 from .utils import cache_id, entity_id, status_to_exception
 
 
@@ -471,9 +473,9 @@ def sql(
         elif isinstance(cache, AioCache):
             c_info = cache.cache_info
         else:
-            c_info = None
+            c_info = CacheInfo(protocol_context=self.protocol_context)
 
-        if c_info:
+        if c_info.cache_id:
             schema = None
 
         return AioSqlFieldsCursor(self, c_info, query_str, page_size, query_args, schema, statement_type,
@@ -487,3 +489,21 @@ def get_cluster(self) -> 'AioCluster':
         :return: :py:class:`~pyignite.aio_cluster.AioCluster` instance.
         """
         return AioCluster(self)
+
+    def tx_start(self, concurrency: TransactionConcurrency = TransactionConcurrency.PESSIMISTIC,
+                 isolation: TransactionIsolation = TransactionIsolation.REPEATABLE_READ,
+                 timeout: Union[int, float] = 0, label: Optional[str] = None) -> 'AioTransaction':
+        """
+        Start async thin client transaction.
+
+        :param concurrency: (optional) transaction concurrency, see
+                :py:class:`~pyignite.datatypes.transactions.TransactionConcurrency`
+        :param isolation: (optional) transaction isolation level, see
+                :py:class:`~pyignite.datatypes.transactions.TransactionIsolation`
+        :param timeout: (optional) transaction timeout in seconds if float, in millis if int
+        :param label: (optional) transaction label.
+        :return: :py:class:`~pyignite.transaction.AioTransaction` instance.
+        """
+        if sys.version_info < (3, 7):
+            raise NotSupportedError(f"Transactions are not supported in async client on current python {sys.version}")
+        return AioTransaction(self, concurrency, isolation, timeout, label)

pyignite/api/affinity.py

@@ -68,27 +68,23 @@
 ])
 
 
-def cache_get_node_partitions(conn: 'Connection', caches: Union[int, Iterable[int]], query_id: int = None) -> APIResult:
+def cache_get_node_partitions(conn: 'Connection', caches: Union[int, Iterable[int]]) -> APIResult:
     """
     Gets partition mapping for an Ignite cache or a number of caches. See
     “IEP-23: Best Effort Affinity for thin clients”.
 
     :param conn: connection to Ignite server,
-    :param caches: cache ID(s) the mapping is provided for,
-    :param query_id: (optional) a value generated by client and returned as-is
-     in response.query_id. When the parameter is omitted, a random value
-     is generated,
+    :param caches: cache ID(s) the mapping is provided for
     :return: API result data object.
     """
-    return __cache_get_node_partitions(conn, caches, query_id)
+    return __cache_get_node_partitions(conn, caches)
 
 
-async def cache_get_node_partitions_async(conn: 'AioConnection', caches: Union[int, Iterable[int]],
-                                          query_id: int = None) -> APIResult:
+async def cache_get_node_partitions_async(conn: 'AioConnection', caches: Union[int, Iterable[int]]) -> APIResult:
     """
     Async version of cache_get_node_partitions.
     """
-    return await __cache_get_node_partitions(conn, caches, query_id)
+    return await __cache_get_node_partitions(conn, caches)
 
 
 def __post_process_partitions(result):
@@ -135,13 +131,12 @@ def __post_process_partitions(result):
     return result
 
 
-def __cache_get_node_partitions(conn, caches, query_id):
+def __cache_get_node_partitions(conn, caches):
     query_struct = Query(
         OP_CACHE_PARTITIONS,
         [
             ('cache_ids', cache_ids),
-        ],
-        query_id=query_id
+        ]
     )
     if not is_iterable(caches):
         caches = [caches]

pyignite/api/binary.py

@@ -26,34 +26,30 @@
 from ..queries.response import BinaryTypeResponse
 
 
-def get_binary_type(conn: 'Connection', binary_type: Union[str, int], query_id=None) -> APIResult:
+def get_binary_type(conn: 'Connection', binary_type: Union[str, int]) -> APIResult:
     """
     Gets the binary type information by type ID.
 
     :param conn: connection to Ignite server,
     :param binary_type: binary type name or ID,
-    :param query_id: (optional) a value generated by client and returned as-is
-     in response.query_id. When the parameter is omitted, a random value
-     is generated,
     :return: API result data object.
     """
-    return __get_binary_type(conn, binary_type, query_id)
+    return __get_binary_type(conn, binary_type)
 
 
-async def get_binary_type_async(conn: 'AioConnection', binary_type: Union[str, int], query_id=None) -> APIResult:
+async def get_binary_type_async(conn: 'AioConnection', binary_type: Union[str, int]) -> APIResult:
     """
     Async version of get_binary_type.
     """
-    return await __get_binary_type(conn, binary_type, query_id)
+    return await __get_binary_type(conn, binary_type)
 
 
-def __get_binary_type(conn, binary_type, query_id):
+def __get_binary_type(conn, binary_type):
     query_struct = Query(
         OP_GET_BINARY_TYPE,
         [
             ('type_id', Int),
         ],
-        query_id=query_id,
         response_type=BinaryTypeResponse
     )
 
@@ -63,7 +59,7 @@ def __get_binary_type(conn, binary_type, query_id):
 
 
 def put_binary_type(connection: 'Connection', type_name: str, affinity_key_field: str = None,
-                    is_enum=False, schema: dict = None, query_id=None) -> APIResult:
+                    is_enum=False, schema: dict = None) -> APIResult:
     """
     Registers binary type information in cluster.
 
@@ -76,20 +72,17 @@ def put_binary_type(connection: 'Connection', type_name: str, affinity_key_field
      parameter names as keys and an integers as values. When register binary
      type, pass a dict of field names: field types. Binary type with no fields
      is OK,
-    :param query_id: (optional) a value generated by client and returned as-is
-     in response.query_id. When the parameter is omitted, a random value
-     is generated,
     :return: API result data object.
     """
-    return __put_binary_type(connection, type_name, affinity_key_field, is_enum, schema, query_id)
+    return __put_binary_type(connection, type_name, affinity_key_field, is_enum, schema)
 
 
 async def put_binary_type_async(connection: 'AioConnection', type_name: str, affinity_key_field: str = None,
                                 is_enum=False, schema: dict = None, query_id=None) -> APIResult:
     """
     Async version of put_binary_type.
     """
-    return await __put_binary_type(connection, type_name, affinity_key_field, is_enum, schema, query_id)
+    return await __put_binary_type(connection, type_name, affinity_key_field, is_enum, schema)
 
 
 def __post_process_put_binary(type_id):
@@ -103,7 +96,7 @@ def internal(result):
     return internal
 
 
-def __put_binary_type(connection, type_name, affinity_key_field, is_enum, schema, query_id):
+def __put_binary_type(connection, type_name, affinity_key_field, is_enum, schema):
     # prepare data
     if schema is None:
         schema = {}
@@ -158,8 +151,7 @@ def __put_binary_type(connection, type_name, affinity_key_field, is_enum, schema
                 ('is_enum', Bool),
                 ('enums', enum_struct),
                 ('schema', schema_struct),
-            ],
-            query_id=query_id,
+            ]
         )
     else:
         query_struct = Query(
@@ -171,8 +163,7 @@ def __put_binary_type(connection, type_name, affinity_key_field, is_enum, schema
                 ('binary_fields', binary_fields_struct),
                 ('is_enum', Bool),
                 ('schema', schema_struct),
-            ],
-            query_id=query_id,
+            ]
         )
     return query_perform(query_struct, connection, query_params=data,
                          post_process_fun=__post_process_put_binary(type_id))