-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
See also appsignal/appsignal-nodejs#1079. In Ruby and Node.js, we use `const_missing` and `Proxy` respectively to make `Heartbeat` pretend to be `Cron`, allowing the result of instantiating one to pass an instance class check for the other and vice-versa, while also emitting a deprecation warning in the process. Since the above is not possible in Python, we instead make a `Heartbeat` class that emits the deprecation warning, returns instances of `Cron` when initialised, and also pretends to be `Cron` when instance-checked against. It's not perfect (neither is `Proxy`) but it does the trick.
- Loading branch information
Showing
7 changed files
with
476 additions
and
240 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 | ||
|
||
cron("do_something", do_something) | ||
``` | ||
|
||
Calls to `appsignal.heartbeat` and to the `appsignal.Heartbeat` constructor will emit a deprecation warning. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,87 +1,64 @@ | ||
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 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 _warn_logger_and_stdout(msg: str) -> None: | ||
logger.warning(msg) | ||
print(f"appsignal WARNING: {msg}") | ||
|
||
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() | ||
_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.", | ||
) | ||
|
||
if not config.is_active(): | ||
logger.debug("AppSignal not active, not transmitting heartbeat event") | ||
return | ||
_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.", | ||
) | ||
|
||
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")) | ||
|
||
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 heartbeat(name: str, fn: Callable[[], T] | None = None) -> None | T: | ||
_heartbeat_helper_warning() | ||
return cron(name, fn) | ||
|
||
|
||
def heartbeat(name: str, fn: Callable[[], T] | None = None) -> None | T: | ||
heartbeat = Heartbeat(name) | ||
output = None | ||
class _MetaHeartbeat(type): | ||
def __instancecheck__(cls, other: Any) -> bool: | ||
_heartbeat_class_warning() | ||
return isinstance(other, Cron) | ||
|
||
if fn is not None: | ||
heartbeat.start() | ||
output = fn() | ||
|
||
heartbeat.finish() | ||
return output | ||
class Heartbeat(metaclass=_MetaHeartbeat): | ||
def __new__(cls, name: str) -> Cron: # type: ignore[misc] | ||
_heartbeat_class_warning() | ||
return Cron(name) |
Oops, something went wrong.