Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/openstatusHQ/sdk-python. Official Python SDK for openstatus
Something went wrong. Try again.
4.5 kB · 111 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112"""Sync port of the OpenStatus Node SDK `example.ts`.
Run with:
uv run python examples/basic.py"""
from __future__ import annotations
import datetime as dtimport osimport sys
from openstatus import OpenstatusClientfrom openstatus._gen.openstatus.health.v1.health_pb2 import CheckRequestfrom openstatus._gen.openstatus.monitor.v1.monitor_pb2 import Regionfrom openstatus._gen.openstatus.monitor.v1.service_pb2 import ( GetMonitorHTTPResponseLogRequest, HTTPResponseLogRequestStatus, HTTPResponseLogTrigger, ListMonitorHTTPResponseLogsRequest, ListMonitorsRequest,)
def main() -> None: print("OpenStatus SDK Example\n")
with OpenstatusClient() as client: print("1. Checking API health...") health = client.health.v1.HealthService.check(CheckRequest()) print(f" Status: {health.status}\n")
api_key = os.environ.get("OPENSTATUS_API_KEY") if not api_key: print("2. Skipping monitor operations (OPENSTATUS_API_KEY not set)") print(" Set OPENSTATUS_API_KEY environment variable to test monitor operations.") return
print("2. Listing monitors...") monitors = client.monitor.v1.MonitorService.list_monitors(ListMonitorsRequest()) print(f" Found {monitors.total_size} monitors:") print(f" - HTTP: {len(monitors.http_monitors)}") print(f" - TCP: {len(monitors.tcp_monitors)}") print(f" - DNS: {len(monitors.dns_monitors)}\n")
if monitors.http_monitors: print("3. HTTP Monitors:") for monitor in monitors.http_monitors: state = "active" if monitor.active else "paused" print(f" - {monitor.name}: {monitor.url} ({state})") print()
if not monitors.http_monitors: print("No HTTP monitors found; exiting.") return
first_monitor_id = monitors.http_monitors[0].id print(f"4. Listing HTTP response logs for monitor {first_monitor_id}...") logs = client.monitor.v1.MonitorService.list_monitor_http_response_logs( ListMonitorHTTPResponseLogsRequest(id=first_monitor_id, limit=10) ) has_more = logs.pagination.has_more if logs.HasField("pagination") else False print(f" Found {len(logs.logs)} logs (has_more={has_more})") for log in logs.logs: ts = dt.datetime.fromtimestamp(log.timestamp / 1000, tz=dt.timezone.utc).isoformat() status_code = log.status_code if log.HasField("status_code") else "—" region = Region.Name(log.region) request_status = HTTPResponseLogRequestStatus.Name(log.request_status) trigger = HTTPResponseLogTrigger.Name(log.trigger) log_id = log.id if log.HasField("id") else "—" print( f" {ts} {status_code} {log.latency}ms {region} " f"{request_status} {trigger} id={log_id}" ) if log.HasField("timing"): t = log.timing print( f" timing: dns={t.dns}ms connect={t.connect}ms " f"tls={t.tls}ms ttfb={t.ttfb}ms transfer={t.transfer}ms" )
if not logs.logs or not logs.logs[0].HasField("id"): return
first_log_id = logs.logs[0].id print(f"\n5. Getting log detail for {first_log_id}...") response = client.monitor.v1.MonitorService.get_monitor_http_response_log( GetMonitorHTTPResponseLogRequest(id=first_monitor_id, log_id=first_log_id) ) if response.HasField("log"): detail = response.log print(f" URL: {detail.url}") print(f" Errored: {detail.error}") print(f" Message: {detail.message if detail.HasField('message') else '—'}") inner = detail.log status = inner.status_code if inner.HasField("status_code") else "—" print(f" Status: {status}") assertions = detail.assertions if detail.HasField("assertions") else "—" print(f" Assertions: {assertions}") if detail.headers: print(" Headers:") for header_key, header_value in detail.headers.items(): print(f" {header_key}: {header_value}")
if __name__ == "__main__": try: main() except KeyboardInterrupt: sys.exit(130)