add ttl option to DriftClient.walk method

pkg/drift_client/drift_client.py

@@ -218,6 +218,7 @@ def walk(
         topic: str,
         start: Union[float, datetime, str],
         stop: Union[float, datetime, str],
+        **kwargs,
     ) -> Iterator[DriftDataPackage]:
         """Walks through history data for selected topic
 
@@ -227,7 +228,8 @@ def walk(
                 Format: ISO string, datetime or float timestamp
             stop: End of request timeframe,
                 Format: ISO string, datetime or float timestamp
-
+        KwArgs:
+            ttl: Time to live for the query only for ReductStore
         Returns:
             Iterator with DriftDataPackage
         Raises:
@@ -247,7 +249,7 @@ def walk(
         else:
             start = _convert_type(start)
             stop = _convert_type(stop)
-            for package in self._blob_storage.walk(topic, start, stop):
+            for package in self._blob_storage.walk(topic, start, stop, **kwargs):
                 yield DriftDataPackage(package)
 
     def subscribe_data(self, topic: str, handler: Callable[[DriftDataPackage], None]):

pkg/drift_client/reduct_client.py

@@ -70,20 +70,23 @@ def fetch_data(self, path: str) -> Optional[bytes]:
         except ReductError as err:
             raise DriftClientError(f"Could not read item at {path}") from err
 
-    def walk(self, entry: str, start: int, stop: int) -> Iterator[bytes]:
+    def walk(self, entry: str, start: int, stop: int, **kwargs) -> Iterator[bytes]:
         """
         Walk through the records of an entry between start and stop.
         Args:
             entry: entry name
             start: start timestamp UNIX in seconds
             stop: stop timestamp UNIX in seconds
+        Keyword Args:
+            ttl: time to live for the query
         Raises:
             DriftClientError: if failed to fetch data
         """
 
         bucket: Bucket = self._run(self._client.get_bucket(self._bucket))
 
-        ait = bucket.query(entry, start * 1000_000, stop * 1000_000, ttl=60)
+        ttl = kwargs.get("ttl", 60)
+        ait = bucket.query(entry, start * 1000_000, stop * 1000_000, ttl=ttl)
 
         async def get_next():
             try: