feat(remaining-validity): adds get_secured_api_key_remaining_validity

algoliasearch/exceptions.py

@@ -27,3 +27,7 @@ class AlgoliaUnreachableHostException(AlgoliaException):
 
 class ObjectNotFoundException(AlgoliaException):
     pass
+
+
+class ValidUntilNotFoundException(AlgoliaException):
+    pass

algoliasearch/search_client.py

@@ -1,9 +1,12 @@
 import base64
 import hashlib
 import hmac
+import re
+import time
 
 from typing import Optional, Union, List
 
+from algoliasearch.exceptions import ValidUntilNotFoundException
 from algoliasearch.helpers import endpoint, is_async_available
 from algoliasearch.http.request_options import RequestOptions
 from algoliasearch.http.serializer import QueryParametersSerializer
@@ -302,6 +305,21 @@ def generate_secured_api_key(parent_api_key, restrictions):
 
         return str(base64encoded.decode('utf-8'))
 
+    @staticmethod
+    def get_secured_api_key_remaining_validity(api_key):
+        # type: (str) -> int
+
+        decoded_string = base64.b64decode(api_key)
+
+        match = re.search(r'validUntil=(\d+)', str(decoded_string))
+
+        if match is None:
+            raise ValidUntilNotFoundException(
+                'ValidUntil not found in api key.'
+            )
+
+        return int(match.group(1)) - int(round(time.time()))
+
     def list_indices(self, request_options=None):
         # type: (Optional[Union[dict, RequestOptions]]) -> dict
 

tests/features/test_search_client.py

@@ -5,7 +5,8 @@
 import unittest
 
 from algoliasearch.configs import SearchConfig
-from algoliasearch.exceptions import RequestException
+from algoliasearch.exceptions import RequestException, \
+    ValidUntilNotFoundException
 from algoliasearch.http.hosts import HostsCollection, Host, CallType
 from algoliasearch.http.serializer import QueryParametersSerializer
 from algoliasearch.responses import MultipleResponse
@@ -346,6 +347,34 @@ def test_secured_api_keys(self):
             F.search_client(api_key=api_key).init_index(
                 self.index2.name).search('')
 
+    def test_get_secured_api_key_remaining_validity(self):
+        import time
+
+        now = int(time.time())
+        api_key = SearchClient.generate_secured_api_key('foo', {
+            'validUntil': now - (60 * 10)
+        })
+
+        remaining = SearchClient.get_secured_api_key_remaining_validity(
+            api_key
+        )
+
+        self.assertTrue(remaining < 0)
+
+        api_key = SearchClient.generate_secured_api_key('foo', {
+            'validUntil': now + (60 * 10),
+        })
+
+        remaining = SearchClient.get_secured_api_key_remaining_validity(
+            api_key
+        )
+        self.assertTrue(remaining > 0)
+
+        api_key = SearchClient.generate_secured_api_key('foo', {})
+
+        with self.assertRaises(ValidUntilNotFoundException) as _:
+            SearchClient.get_secured_api_key_remaining_validity(api_key)
+
     @unittest.skipIf(Env.is_community(),
                      "Community can not test personalization operations")
     def test_personalization_strategy(self):

tests/unit/test_search_index.py

@@ -40,6 +40,8 @@ def test_exists(self):
 
             self.index.get_settings.assert_called_once()
 
+            self.assertEqual(indexExists, False)
+
             # No request options
             args = self.index.get_settings.call_args[0]
             self.assertEqual(args[0], None)