pinexq-tools

Utility tools for the DataCybernetics PinexQ platform. Distributed as pinexq-tools; modules live under the pinexq namespace.

Modules

pinexq.monitoring

Context managers that report an UptimeKuma push heartbeat when a block finishes — status=up on clean exit, status=down on exception — with ping set to the time spent inside the context.

from pinexq.monitoring import UptimeKumaPush, AsyncUptimeKumaPush

with UptimeKumaPush("https://kuma.example/api/push/abc123"):
    do_work()

async with AsyncUptimeKumaPush("https://kuma.example/api/push/abc123"):
    await do_work()

Read-only helpers that ask a Prometheus HTTP API whether a service's alert rules are active, per environment (distinguished by the tenant_cluster_name label). Instead of re-deriving health from raw metrics, they read the alerts the rules already encode.

query_prometheus_alerts_absent(...) returns True when there are no pending/firing alerts for the service on that cluster (i.e. healthy):

from pinexq.monitoring import query_prometheus_alerts_absent

healthy = query_prometheus_alerts_absent(
    "https://prometheus.example",   # base URL, without the /api/v1 suffix
    "epex-trader",                  # value of the `service` label on the rules
    "pinexq-k3s-prod",              # value of the `tenant_cluster_name` label
    username="user",
    password="password",
)
# healthy is True when the service has no active alerts on that cluster

query_prometheus_alert_rules(...) returns the per-cluster state of each alert rule for the service as a {rule_name: state} mapping ("inactive", "pending", or "firing"). Because it enumerates loaded rules, it also confirms the rules are present:

from pinexq.monitoring import query_prometheus_alert_rules

states = query_prometheus_alert_rules(
    "https://prometheus.example",
    "epex-trader",
    "pinexq-k3s-prod",
    username="user",
    password="password",
)
# {"M7MessagesDropped": "inactive", "M7QueueSpilling": "firing", ...}
all_ok = all(state == "inactive" for state in states.values())

all_prometheus_alerts_inactive(...) wraps that call and returns a single boolean — True when every matching rule is inactive. Unlike query_prometheus_alerts_absent, it raises ValueError if the service / tenant_cluster_name matched no rules, so a typo can't be mistaken for a healthy result:

from pinexq.monitoring import all_prometheus_alerts_inactive

healthy = all_prometheus_alerts_inactive(
    "https://prometheus.example",
    "epex-trader",
    "pinexq-k3s-prod",
    username="user",
    password="password",
)

Credentials may instead be embedded in the connection string as userinfo — explicit username / password override any embedded ones:

query_prometheus_alerts_absent(
    "https://user:password@prometheus.example", "epex-trader", "pinexq-k3s-prod"
)

Both accept an optional keyword-only timeout (seconds, default 10.0), authenticate via HTTP basic auth, and raise httpx.HTTPStatusError on a non-2xx response or RuntimeError if the API reports a non-success status.

Development

uv sync --all-groups
uv build
ruff check .
pytest