Skip to content

Monitoring

erpc.monitoring

Health monitoring for eRPC upstream endpoints.

HealthStatus dataclass

Structured result from an eRPC health check.

Attributes:

Name Type Description
status str

Health status string (e.g. "ok").
uptime float

Process uptime in seconds.
version str

eRPC version string.
Source code in erpc/client.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@dataclass(frozen=True)
class HealthStatus:
    """Structured result from an eRPC health check.

    Attributes:
        status: Health status string (e.g. ``"ok"``).
        uptime: Process uptime in seconds.
        version: eRPC version string.

    """

    status: str
    uptime: float
    version: str

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> HealthStatus:
        """Create a ``HealthStatus`` from a JSON-decoded dictionary.

        Missing fields are filled with sensible defaults so partial
        responses don't crash the client.

        Args:
            data: Decoded JSON body from the health endpoint.

        Returns:
            Populated ``HealthStatus`` instance.

        """
        return cls(
            status=str(data.get("status", "unknown")),
            uptime=float(data.get("uptime", 0.0)),
            version=str(data.get("version", "unknown")),
        )

from_dict classmethod 

from_dict(data: dict[str, Any]) -> HealthStatus

Create a HealthStatus from a JSON-decoded dictionary.

Missing fields are filled with sensible defaults so partial responses don't crash the client.

Parameters:

Name Type Description Default
data dict[str, Any]

Decoded JSON body from the health endpoint.

 required

Returns:

Type Description
HealthStatus

Populated HealthStatus instance.
Source code in erpc/client.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@classmethod
def from_dict(cls, data: dict[str, Any]) -> HealthStatus:
    """Create a ``HealthStatus`` from a JSON-decoded dictionary.

    Missing fields are filled with sensible defaults so partial
    responses don't crash the client.

    Args:
        data: Decoded JSON body from the health endpoint.

    Returns:
        Populated ``HealthStatus`` instance.

    """
    return cls(
        status=str(data.get("status", "unknown")),
        uptime=float(data.get("uptime", 0.0)),
        version=str(data.get("version", "unknown")),
    )

HealthEvent

Bases: Enum

Events emitted by the health monitor.

Source code in erpc/monitoring.py 
11
12
13
14
15
16
17
class HealthEvent(enum.Enum):
    """Events emitted by the health monitor."""

    HEALTHY = "healthy"
    DEGRADED = "degraded"
    DOWN = "down"
    RECOVERED = "recovered"

HealthMonitor dataclass

Monitors health of eRPC upstream endpoints.

Attributes:

Name Type Description
url str

Base URL of the eRPC instance to monitor.
interval float

Polling interval in seconds.
history list[HealthStatus]

Recent health check results.
Source code in erpc/monitoring.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@dataclass
class HealthMonitor:
    """Monitors health of eRPC upstream endpoints.

    Attributes:
        url: Base URL of the eRPC instance to monitor.
        interval: Polling interval in seconds.
        history: Recent health check results.

    """

    url: str = "http://127.0.0.1:4000"
    interval: float = 30.0
    history: list[HealthStatus] = field(default_factory=list)

    def latest_event(self) -> HealthEvent | None:
        """Return the latest health event based on history, or ``None``."""
        if not self.history:
            return None
        last = self.history[-1]
        if last.status == "ok":
            return HealthEvent.HEALTHY
        return HealthEvent.DOWN

latest_event 

latest_event() -> HealthEvent | None

Return the latest health event based on history, or None.

Source code in erpc/monitoring.py 
35
36
37
38
39
40
41
42
def latest_event(self) -> HealthEvent | None:
    """Return the latest health event based on history, or ``None``."""
    if not self.history:
        return None
    last = self.history[-1]
    if last.status == "ok":
        return HealthEvent.HEALTHY
    return HealthEvent.DOWN