add min_ping_interval to ping

Author: snopoke

taskbadger/sdk.py

@@ -1,3 +1,4 @@
+import datetime
 import logging
 import os
 from typing import Any
@@ -392,9 +393,21 @@ def tag(self, tags: dict[str, str]):
         """Add tags to the task."""
         self.update(tags=tags)
 
-    def ping(self):
+    def ping(self, min_ping_interval=None):
         """Update the task without changing any values. This can be used in conjunction
-        with 'stale_timeout' to indicate that the task is still running."""
+        with 'stale_timeout' to indicate that the task is still running.
+
+        Arguments:
+            min_ping_interval: The minimum interval between pings in seconds. If set this will only
+                update the task if the last update was more than `min_ping_interval` seconds ago.
+        """
+        if min_ping_interval and self._task.updated:
+            # tzinfo should always be set but for the sake of safety we check
+            tz = None if self._task.updated.tzinfo is None else datetime.UTC
+            now = datetime.datetime.now(tz)
+            time_since = now - self._task.updated
+            if time_since.total_seconds() < min_ping_interval:
+                return
         self.update()
 
     @property

tests/test_sdk.py

@@ -1,3 +1,4 @@
+import datetime
 from http import HTTPStatus
 from unittest import mock
 
@@ -153,6 +154,26 @@ def test_update_data(settings, patched_update):
     _verify_update(settings, patched_update, data={"a": 1})
 
 
+def test_ping(settings, patched_update):
+    task = Task(task_for_test())
+
+    updated_at = task.updated
+    patched_update.return_value = Response(HTTPStatus.OK, b"", {}, task_for_test())
+    task.ping(min_ping_interval=1)
+    assert len(patched_update.call_args_list) == 0
+
+    task.ping()
+    _verify_update(settings, patched_update)
+    assert task.updated > updated_at
+
+    task.ping(min_ping_interval=1)
+    assert len(patched_update.call_args_list) == 1
+
+    task._task.updated = task._task.updated - datetime.timedelta(seconds=1)
+    task.ping(min_ping_interval=1)
+    assert len(patched_update.call_args_list) == 2
+
+
 def test_increment_progress(settings, patched_update):
     api_task = task_for_test()
     task = Task(api_task)

tests/utils.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+import datetime
 from uuid import uuid4
 
 from taskbadger.internal.models import Task as TaskInternal
@@ -12,8 +12,8 @@ def task_for_test(**kwargs):
     kwargs["url"] = None
     kwargs["public_url"] = None
     kwargs["value_percent"] = None
-    kwargs["created"] = datetime.utcnow()
-    kwargs["updated"] = datetime.utcnow()
+    kwargs["created"] = datetime.datetime.now(datetime.UTC)
+    kwargs["updated"] = datetime.datetime.now(datetime.UTC)
     return TaskInternal(
         task_id,
         "org",