GG-33908 IGNITE-14911 Unify timeouts, add support for datetime.timedelta for expiry_policy (#52)

(cherry picked from commit 92a115c)

Co-authored-by: Ivan Daschinsky <[email protected]>

Author: isapego

docs/async_examples.rst

@@ -62,20 +62,20 @@ in cache settings dictionary on creation.
 .. literalinclude:: ../examples/expiry_policy.py
   :language: python
   :dedent: 12
-  :lines: 72-75
+  :lines: 73-76
 
 .. literalinclude:: ../examples/expiry_policy.py
   :language: python
   :dedent: 12
-  :lines: 81-89
+  :lines: 82-90
 
 Secondly, expiry policy can be set for all cache operations, which are done under decorator. To create it use
 :py:meth:`~pygridgain.cache.BaseCache.with_expire_policy`
 
 .. literalinclude:: ../examples/expiry_policy.py
   :language: python
   :dedent: 12
-  :lines: 96-105
+  :lines: 97-106
 
 Transactions
 ------------

docs/examples.rst

@@ -96,20 +96,20 @@ in cache settings dictionary on creation.
 .. literalinclude:: ../examples/expiry_policy.py
   :language: python
   :dedent: 12
-  :lines: 31-34
+  :lines: 32-35
 
 .. literalinclude:: ../examples/expiry_policy.py
   :language: python
   :dedent: 12
-  :lines: 40-46
+  :lines: 41-47
 
 Secondly, expiry policy can be set for all cache operations, which are done under decorator. To create it use
 :py:meth:`~pygridgain.cache.BaseCache.with_expire_policy`
 
 .. literalinclude:: ../examples/expiry_policy.py
   :language: python
   :dedent: 12
-  :lines: 53-60
+  :lines: 54-61
 
 Scan
 ====

examples/expiry_policy.py

@@ -15,6 +15,7 @@
 #
 import asyncio
 import time
+from datetime import timedelta
 
 from pygridgain import Client, AioClient
 from pygridgain.datatypes import ExpiryPolicy
@@ -31,7 +32,7 @@ def main():
         try:
             ttl_cache = client.create_cache({
                 PROP_NAME: 'test',
-                PROP_EXPIRY_POLICY: ExpiryPolicy(create=1.0)
+                PROP_EXPIRY_POLICY: ExpiryPolicy(create=timedelta(seconds=1.0))
             })
         except NotSupportedByClusterError:
             print("'ExpiryPolicy' API is not supported by cluster. Finishing...")
@@ -51,7 +52,7 @@ def main():
         print("Create simple Cache and set TTL through `with_expire_policy`")
         simple_cache = client.create_cache('test')
         try:
-            ttl_cache = simple_cache.with_expire_policy(access=1.0)
+            ttl_cache = simple_cache.with_expire_policy(access=timedelta(seconds=1.0))
             ttl_cache.put(1, 1)
             time.sleep(0.5)
             print(f"key = {1}, value = {ttl_cache.get(1)}")
@@ -72,7 +73,7 @@ async def async_main():
         try:
             ttl_cache = await client.create_cache({
                 PROP_NAME: 'test',
-                PROP_EXPIRY_POLICY: ExpiryPolicy(create=1.0)
+                PROP_EXPIRY_POLICY: ExpiryPolicy(create=timedelta(seconds=1.0))
             })
         except NotSupportedByClusterError:
             print("'ExpiryPolicy' API is not supported by cluster. Finishing...")
@@ -94,7 +95,7 @@ async def async_main():
         print("Create simple Cache and set TTL through `with_expire_policy`")
         simple_cache = await client.create_cache('test')
         try:
-            ttl_cache = simple_cache.with_expire_policy(access=1.0)
+            ttl_cache = simple_cache.with_expire_policy(access=timedelta(seconds=1.0))
             await ttl_cache.put(1, 1)
             await asyncio.sleep(0.5)
             value = await ttl_cache.get(1)

examples/transactions.py

@@ -62,7 +62,7 @@ async def async_example():
 
         # rollback transaction on timeout.
         try:
-            async with client.tx_start(timeout=1.0, label='long-tx') as tx:
+            async with client.tx_start(timeout=1000, label='long-tx') as tx:
                 await cache.put(key, 'fail')
                 await asyncio.sleep(2.0)
                 await tx.commit()
@@ -114,7 +114,7 @@ def sync_example():
 
         # rollback transaction on timeout.
         try:
-            with client.tx_start(timeout=1.0, label='long-tx') as tx:
+            with client.tx_start(timeout=1000, label='long-tx') as tx:
                 cache.put(key, 'fail')
                 time.sleep(2.0)
                 tx.commit()

pygridgain/aio_client.py

@@ -490,15 +490,15 @@ def get_cluster(self) -> 'AioCluster':
 
     def tx_start(self, concurrency: TransactionConcurrency = TransactionConcurrency.PESSIMISTIC,
                  isolation: TransactionIsolation = TransactionIsolation.REPEATABLE_READ,
-                 timeout: Union[int, float] = 0, label: Optional[str] = None) -> 'AioTransaction':
+                 timeout: int = 0, label: Optional[str] = None) -> 'AioTransaction':
         """
         Start async thin client transaction. **Supported only python 3.7+**
 
         :param concurrency: (optional) transaction concurrency, see
-                :py:class:`~pygridgain.datatypes.transactions.TransactionConcurrency`
+                :py:class:`~pygridgain.datatypes.transactions.TransactionConcurrency`,
         :param isolation: (optional) transaction isolation level, see
-                :py:class:`~pygridgain.datatypes.transactions.TransactionIsolation`
-        :param timeout: (optional) transaction timeout in seconds if float, in millis if int
+                :py:class:`~pygridgain.datatypes.transactions.TransactionIsolation`,
+        :param timeout: (optional) transaction timeout in milliseconds,
         :param label: (optional) transaction label.
         :return: :py:class:`~pygridgain.transaction.AioTransaction` instance.
         """

pygridgain/cache.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-import time
+import datetime
 from typing import Any, Iterable, Optional, Tuple, Union
 
 from .api.tx_api import get_tx_connection
@@ -137,16 +137,16 @@ def cache_id(self) -> int:
 
     def with_expire_policy(
             self, expiry_policy: Optional[ExpiryPolicy] = None,
-            create: Union[int, float] = ExpiryPolicy.UNCHANGED,
-            update: Union[int, float] = ExpiryPolicy.UNCHANGED,
-            access: Union[int, float] = ExpiryPolicy.UNCHANGED
+            create: Union[int, datetime.timedelta] = ExpiryPolicy.UNCHANGED,
+            update: Union[int, datetime.timedelta] = ExpiryPolicy.UNCHANGED,
+            access: Union[int, datetime.timedelta] = ExpiryPolicy.UNCHANGED
     ):
         """
         :param expiry_policy: optional :class:`~pygridgain.datatypes.expiry_policy.ExpiryPolicy`
-         object. If it is set, other params will be ignored.
-        :param create: create TTL in seconds (float) or milliseconds (int),
-        :param update: Create TTL in seconds (float) or milliseconds (int),
-        :param access: Create TTL in seconds (float) or milliseconds (int).
+         object. If it is set, other params will be ignored,
+        :param create: TTL for create in milliseconds or :py:class:`~time.timedelta`,
+        :param update: TTL for update in milliseconds or :py:class:`~time.timedelta`,
+        :param access: TTL for access in milliseconds or :py:class:`~time.timedelta`,
         :return: cache decorator with expiry policy set.
         """
         if not self.client.protocol_context.is_expiry_policy_supported():

pygridgain/client.py

@@ -745,15 +745,15 @@ def get_cluster(self) -> 'Cluster':
 
     def tx_start(self, concurrency: TransactionConcurrency = TransactionConcurrency.PESSIMISTIC,
                  isolation: TransactionIsolation = TransactionIsolation.REPEATABLE_READ,
-                 timeout: Union[int, float] = 0, label: Optional[str] = None) -> 'Transaction':
+                 timeout: int = 0, label: Optional[str] = None) -> 'Transaction':
         """
         Start thin client transaction.
 
         :param concurrency: (optional) transaction concurrency, see
-                :py:class:`~pygridgain.datatypes.transactions.TransactionConcurrency`
+                :py:class:`~pygridgain.datatypes.transactions.TransactionConcurrency`,
         :param isolation: (optional) transaction isolation level, see
-                :py:class:`~pygridgain.datatypes.transactions.TransactionIsolation`
-        :param timeout: (optional) transaction timeout in seconds if float, in millis if int
+                :py:class:`~pygridgain.datatypes.transactions.TransactionIsolation`,
+        :param timeout: (optional) transaction timeout in milliseconds,
         :param label: (optional) transaction label.
         :return: :py:class:`~pygridgain.transaction.Transaction` instance.
         """

pygridgain/datatypes/cache_properties.py

@@ -14,6 +14,8 @@
 # limitations under the License.
 #
 import ctypes
+import math
+from typing import Union
 
 from . import ExpiryPolicy
 from .prop_codes import *
@@ -137,6 +139,20 @@ async def from_python_async(cls, stream, value):
         return cls.from_python(stream, value)
 
 
+class TimeoutProp(PropBase):
+    prop_data_class = Long
+
+    @classmethod
+    def from_python(cls, stream, value: int):
+        if not isinstance(value, int) or value < 0:
+            raise ValueError(f'Timeout value should be a positive integer, {value} passed instead')
+        return super().from_python(stream, value)
+
+    @classmethod
+    async def from_python_async(cls, stream, value):
+        return cls.from_python(stream, value)
+
+
 class PropName(PropBase):
     prop_code = PROP_NAME
     prop_data_class = String
@@ -227,9 +243,8 @@ class PropRebalanceDelay(PropBase):
     prop_data_class = Long
 
 
-class PropRebalanceTimeout(PropBase):
+class PropRebalanceTimeout(TimeoutProp):
     prop_code = PROP_REBALANCE_TIMEOUT
-    prop_data_class = Long
 
 
 class PropRebalanceBatchSize(PropBase):
@@ -262,9 +277,8 @@ class PropCacheKeyConfiguration(PropBase):
     prop_data_class = CacheKeyConfiguration
 
 
-class PropDefaultLockTimeout(PropBase):
+class PropDefaultLockTimeout(TimeoutProp):
     prop_code = PROP_DEFAULT_LOCK_TIMEOUT
-    prop_data_class = Long
 
 
 class PropMaxConcurrentAsyncOperation(PropBase):

pygridgain/datatypes/expiry_policy.py

@@ -14,6 +14,8 @@
 # limitations under the License.
 #
 import ctypes
+import math
+from datetime import timedelta
 from io import SEEK_CUR
 from typing import Union
 
@@ -23,13 +25,16 @@
 
 
 def _positive(_, attrib, value):
+    if isinstance(value, timedelta):
+        value = value.total_seconds() * 1000
+
     if value < 0 and value not in [ExpiryPolicy.UNCHANGED, ExpiryPolicy.ETERNAL]:
         raise ValueError(f"'{attrib.name}' value must not be negative")
 
 
 def _write_duration(stream, value):
-    if isinstance(value, float):
-        value = int(value * 1000)
+    if isinstance(value, timedelta):
+        value = math.floor(value.total_seconds() * 1000)
 
     stream.write(value.to_bytes(8, byteorder=PROTOCOL_BYTE_ORDER, signed=True))
 
@@ -45,17 +50,17 @@ class ExpiryPolicy:
     #: Set TTL eternal.
     ETERNAL = -1
 
-    #: Set TTL for create in seconds(float) or millis(int)
-    create = attr.ib(kw_only=True, default=UNCHANGED,
-                     validator=[attr.validators.instance_of((int, float)), _positive])
+    #: Set TTL for create in milliseconds or :py:class:`~time.timedelta`
+    create = attr.ib(kw_only=True, default=UNCHANGED, type=Union[int, timedelta],
+                     validator=[attr.validators.instance_of((int, timedelta)), _positive])
 
-    #: Set TTL for update in seconds(float) or millis(int)
-    update = attr.ib(kw_only=True, default=UNCHANGED, type=Union[int, float],
-                     validator=[attr.validators.instance_of((int, float)), _positive])
+    #: Set TTL for update in milliseconds or :py:class:`~time.timedelta`
+    update = attr.ib(kw_only=True, default=UNCHANGED, type=Union[int, timedelta],
+                     validator=[attr.validators.instance_of((int, timedelta)), _positive])
 
-    #: Set TTL for access in seconds(float) or millis(int)
-    access = attr.ib(kw_only=True, default=UNCHANGED, type=Union[int, float],
-                     validator=[attr.validators.instance_of((int, float)), _positive])
+    #: Set TTL for access in milliseconds or :py:class:`~time.timedelta`
+    access = attr.ib(kw_only=True, default=UNCHANGED, type=Union[int, timedelta],
+                     validator=[attr.validators.instance_of((int, timedelta)), _positive])
 
     class _CType(ctypes.LittleEndianStructure):
         _pack_ = 1

pygridgain/transaction.py

@@ -14,30 +14,50 @@
 # limitations under the License.
 #
 
-import math
-from typing import Union
+from enum import IntEnum
+from typing import Union, Type
 
 from pygridgain.api.tx_api import tx_end, tx_start, tx_end_async, tx_start_async
 from pygridgain.datatypes import TransactionIsolation, TransactionConcurrency
 from pygridgain.exceptions import CacheError
 from pygridgain.utils import status_to_exception
 
 
-def _convert_to_millis(timeout: Union[int, float]) -> int:
-    if isinstance(timeout, float):
-        return math.floor(timeout * 1000)
-    return timeout
+def _validate_int_enum_param(value: Union[int, IntEnum], cls: Type[IntEnum]):
+    if value not in cls:
+        raise ValueError(f'{value} not in {cls}')
+    return value
 
 
-class Transaction:
+def _validate_timeout(value):
+    if not isinstance(value, int) or value < 0:
+        raise ValueError(f'Timeout value should be a positive integer, {value} passed instead')
+    return value
+
+
+def _validate_label(value):
+    if value and not isinstance(value, str):
+        raise ValueError(f'Label should be str, {type(value)} passed instead')
+    return value
+
+
+class _BaseTransaction:
+    def __init__(self, client, concurrency=TransactionConcurrency.PESSIMISTIC,
+                 isolation=TransactionIsolation.REPEATABLE_READ, timeout=0, label=None):
+        self.client = client
+        self.concurrency = _validate_int_enum_param(concurrency, TransactionConcurrency)
+        self.isolation = _validate_int_enum_param(isolation, TransactionIsolation)
+        self.timeout = _validate_timeout(timeout)
+        self.label, self.closed = _validate_label(label), False
+
+
+class Transaction(_BaseTransaction):
     """
     Thin client transaction.
     """
     def __init__(self, client, concurrency=TransactionConcurrency.PESSIMISTIC,
                  isolation=TransactionIsolation.REPEATABLE_READ, timeout=0, label=None):
-        self.client, self.concurrency = client, concurrency
-        self.isolation, self.timeout = isolation, _convert_to_millis(timeout)
-        self.label, self.closed = label, False
+        super().__init__(client, concurrency, isolation, timeout, label)
         self.tx_id = self.__start_tx()
 
     def commit(self) -> None:
@@ -78,15 +98,13 @@ def __end_tx(self, committed):
         return tx_end(self.tx_id, committed)
 
 
-class AioTransaction:
+class AioTransaction(_BaseTransaction):
     """
     Async thin client transaction.
     """
     def __init__(self, client, concurrency=TransactionConcurrency.PESSIMISTIC,
                  isolation=TransactionIsolation.REPEATABLE_READ, timeout=0, label=None):
-        self.client, self.concurrency = client, concurrency
-        self.isolation, self.timeout = isolation, _convert_to_millis(timeout)
-        self.label, self.closed = label, False
+        super().__init__(client, concurrency, isolation, timeout, label)
 
     def __await__(self):
         return (yield from self.__aenter__().__await__())