Rename heartbeats to cron check-ins

See also appsignal/appsignal-nodejs#1079.

.changesets/rename-heartbeats-to-cron-check-ins.md

@@ -0,0 +1,26 @@
+---
+bump: patch
+type: change
+---
+
+Rename heartbeats to cron check-ins. Calls to `appsignal.heartbeat` and `appsignal.Heartbeat` should be replaced with calls to `appsignal.check_in.cron` and `appsignal.check_in.Cron`, for example:
+
+```python
+# Before
+from appsignal import heartbeat
+
+def do_something():
+  pass
+
+heartbeat("do_something", do_something)
+
+# After
+from appsignal.check_in import cron
+
+def do_something():
+  pass
+
+heartbeat("do_something", do_something)
+```
+
+Calls to `appsignal.heartbeat` and to the `appsignal.Heartbeat` constructor will emit a deprecation warning.

conftest.py

@@ -17,6 +17,7 @@
 from appsignal import probes
 from appsignal.agent import agent
 from appsignal.client import _reset_client
+from appsignal.heartbeat import _heartbeat_class_warning, _heartbeat_helper_warning
 from appsignal.internal_logger import _reset_logger
 from appsignal.opentelemetry import METRICS_PREFERRED_TEMPORALITY
 
@@ -110,3 +111,11 @@ def stop_agent():
     working_dir = os.path.join(tmp_path, "appsignal")
     if os.path.isdir(working_dir):
         os.system(f"rm -rf {working_dir}")
+
+
+@pytest.fixture(scope="function")
+def reset_heartbeat_warnings():
+    _heartbeat_class_warning.reset()
+    _heartbeat_helper_warning.reset()
+
+    yield

src/appsignal/__init__.py

@@ -3,6 +3,7 @@
 import os
 from runpy import run_path
 
+from . import check_in
 from .client import Client as Appsignal
 from .heartbeat import Heartbeat, heartbeat
 from .metrics import add_distribution_value, increment_counter, set_gauge
@@ -45,6 +46,7 @@
     "add_distribution_value",
     "Heartbeat",
     "heartbeat",
+    "check_in",
     "start",
 ]
 

src/appsignal/check_in.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from binascii import hexlify
+from os import urandom
+from time import time
+from typing import Any, Callable, Literal, TypedDict, TypeVar, Union
+
+from . import internal_logger as logger
+from .client import Client
+from .config import Config
+from .transmitter import transmit
+
+
+T = TypeVar("T")
+
+EventKind = Union[Literal["start"], Literal["finish"]]
+
+
+class Event(TypedDict):
+    name: str
+    id: str
+    kind: EventKind
+    timestamp: int
+
+
+class Cron:
+    name: str
+    id: str
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.id = hexlify(urandom(8)).decode("utf-8")
+
+    def _event(self, kind: EventKind) -> Event:
+        return Event(name=self.name, id=self.id, kind=kind, timestamp=int(time()))
+
+    def _transmit(self, event: Event) -> None:
+        config = Client.config() or Config()
+
+        if not config.is_active():
+            logger.debug("AppSignal not active, not transmitting cron check-in event")
+            return
+
+        url = f"{config.option('logging_endpoint')}/checkins/cron/json"
+        try:
+            response = transmit(url, json=event)
+            if 200 <= response.status_code <= 299:
+                logger.debug(
+                    f"Transmitted cron check-in `{event['name']}` ({event['id']}) "
+                    f"{event['kind']} event"
+                )
+            else:
+                logger.error(
+                    f"Failed to transmit cron check-in {event['kind']} event: "
+                    f"status code was {response.status_code}"
+                )
+        except Exception as e:
+            logger.error(f"Failed to transmit cron check-in {event['kind']} event: {e}")
+
+    def start(self) -> None:
+        self._transmit(self._event("start"))
+
+    def finish(self) -> None:
+        self._transmit(self._event("finish"))
+
+    def __enter__(self) -> None:
+        self.start()
+
+    def __exit__(
+        self, exc_type: Any = None, exc_value: Any = None, traceback: Any = None
+    ) -> Literal[False]:
+        if exc_type is None:
+            self.finish()
+
+        return False
+
+
+def cron(name: str, fn: Callable[[], T] | None = None) -> None | T:
+    cron = Cron(name)
+    output = None
+
+    if fn is not None:
+        cron.start()
+        output = fn()
+
+    cron.finish()
+    return output

src/appsignal/heartbeat.py

@@ -1,87 +1,66 @@
 from __future__ import annotations
 
-from binascii import hexlify
-from os import urandom
-from time import time
-from typing import Any, Callable, Literal, TypedDict, TypeVar, Union
+import warnings
+from typing import Any, Callable, TypeVar
 
 from . import internal_logger as logger
-from .client import Client
-from .config import Config
-from .transmitter import transmit
+from .check_in import Cron, cron
 
 
 T = TypeVar("T")
 
-EventKind = Union[Literal["start"], Literal["finish"]]
 
+class _Once:
+    def __init__(self, func: Callable[..., None], *args: Any, **kwargs: Any) -> None:
+        self.called = False
+        self.func = func
+        self.args = args
+        self.kwargs = kwargs
 
-class Event(TypedDict):
-    name: str
-    id: str
-    kind: EventKind
-    timestamp: int
+    def __call__(self) -> None:
+        if not self.called:
+            self.called = True
+            self.func(*self.args, **self.kwargs)
 
+    def reset(self) -> None:
+        self.called = False
 
-class Heartbeat:
-    name: str
-    id: str
-
-    def __init__(self, name: str) -> None:
-        self.name = name
-        self.id = hexlify(urandom(8)).decode("utf-8")
-
-    def _event(self, kind: EventKind) -> Event:
-        return Event(name=self.name, id=self.id, kind=kind, timestamp=int(time()))
-
-    def _transmit(self, event: Event) -> None:
-        config = Client.config() or Config()
 
-        if not config.is_active():
-            logger.debug("AppSignal not active, not transmitting heartbeat event")
-            return
+def _warn_logger_and_stdout(msg: str) -> None:
+    logger.warning(msg)
+    warnings.warn(f"appsignal: {msg}", DeprecationWarning, stacklevel=4)
 
-        url = f"{config.option('logging_endpoint')}/heartbeats/json"
-        try:
-            response = transmit(url, json=event)
-            if 200 <= response.status_code <= 299:
-                logger.debug(
-                    f"Transmitted heartbeat `{event['name']}` ({event['id']}) "
-                    f"{event['kind']} event"
-                )
-            else:
-                logger.error(
-                    "Failed to transmit heartbeat event: "
-                    f"status code was {response.status_code}"
-                )
-        except Exception as e:
-            logger.error(f"Failed to transmit heartbeat event: {e}")
 
-    def start(self) -> None:
-        self._transmit(self._event("start"))
+_heartbeat_helper_warning = _Once(
+    _warn_logger_and_stdout,
+    "The helper `heartbeat` has been deprecated. "
+    "Please update uses of the helper `heartbeat(...)` to `cron(...)`, "
+    "importing it as `from appsignal.check_in import cron`, "
+    "in order to remove this message.",
+)
 
-    def finish(self) -> None:
-        self._transmit(self._event("finish"))
+_heartbeat_class_warning = _Once(
+    _warn_logger_and_stdout,
+    "The class `Heartbeat` has been deprecated. "
+    "Please update uses of the class `Heartbeat(...)` to `Cron(...)` "
+    "importing it as `from appsignal.check_in import Cron`, "
+    "in order to remove this message.",
+)
 
-    def __enter__(self) -> None:
-        self.start()
 
-    def __exit__(
-        self, exc_type: Any = None, exc_value: Any = None, traceback: Any = None
-    ) -> Literal[False]:
-        if exc_type is None:
-            self.finish()
+def heartbeat(name: str, fn: Callable[[], T] | None = None) -> None | T:
+    _heartbeat_helper_warning()
+    return cron(name, fn)
 
-        return False
 
+class _MetaHeartbeat:
+    def __instancecheck__(self, other: Any) -> bool:
+        return isinstance(other, Cron)
 
-def heartbeat(name: str, fn: Callable[[], T] | None = None) -> None | T:
-    heartbeat = Heartbeat(name)
-    output = None
 
-    if fn is not None:
-        heartbeat.start()
-        output = fn()
+class Heartbeat:
+    __metaclass__ = _MetaHeartbeat
 
-    heartbeat.finish()
-    return output
+    def __new__(cls, name: str) -> Cron:  # type: ignore[misc]
+        _heartbeat_class_warning()
+        return Cron(name)