impl/python.py

from typing import Optional

from .monitor_heartbeat_types import HeartbeatStatus


def _whole(name: str, v: object) -> None:
    if not isinstance(v, int) or isinstance(v, bool):
        raise ValueError("%s must be whole seconds, received %r" % (name, v))


def heartbeat_status(last_ping_at: Optional[int], now: int, interval_seconds: int, grace_seconds: int) -> HeartbeatStatus:
    """New, up, late or down, as healthchecks.io names them. Up through the
    due second itself, late through the last second of the grace time, down
    after. A ping stamped in the future (clock skew between the job and the
    monitor) is simply up: its due time is later still."""
    if last_ping_at is not None:
        _whole("lastPingAt", last_ping_at)
    _whole("now", now)
    _whole("intervalSeconds", interval_seconds)
    _whole("graceSeconds", grace_seconds)
    if interval_seconds < 1:
        raise ValueError("intervalSeconds must be at least 1, received %d" % interval_seconds)
    if grace_seconds < 0:
        raise ValueError("graceSeconds must not be negative, received %d" % grace_seconds)
    if last_ping_at is None:
        return HeartbeatStatus(state="new", due_at=None, down_at=None, overdue_seconds=0)
    due_at = last_ping_at + interval_seconds
    down_at = due_at + grace_seconds
    if now <= due_at:
        return HeartbeatStatus(state="up", due_at=due_at, down_at=down_at, overdue_seconds=0)
    return HeartbeatStatus(state="late" if now <= down_at else "down", due_at=due_at, down_at=down_at, overdue_seconds=now - due_at)