OpShift

Python SDK

PreviousNext

opshift — stdlib only, Python 3.9+, mirrors the TypeScript SDK API.

Installation

bash
pip install opshift

Requires Python 3.9 or later. No third-party runtime dependencies.

Initialise the client

python
import os
from opshift import OpShift
 
opshift = OpShift(secret=os.environ["OPSHIFT_SECRET"])

Constructor options

FieldTypeDescription
secretrequiredstrTeam webhook secret (64-char hex). Find it in Settings → Webhook Secret.
base_urlstrOverride the API root. Defaults to https://www.opshift.io.
timeoutfloatPer-request timeout in seconds. Defaults to 30.
transportCallableInject a custom HTTP transport for testing. Signature: (url, body_bytes, headers, timeout) -> (status_code, body_bytes).

opshift.ping(ping_token, *, status, reason, group_key, metadata)

Send a heartbeat ping to a monitor.

python
opshift.ping("payments-worker", status="up", metadata={"batch": 1247})
 
opshift.ping(
    "payments-worker",
    status="down",
    reason="stripe-timeout",
)

Keyword arguments

FieldTypeDescription
status"up" | "down"Defaults to "up" server-side.
reasonstrShort reason string, max 200 chars.
group_keystrAlert grouping key, max 200 chars.
metadatadictArbitrary metadata stored on the ping row.

Return value

python
{
    "message": str,
    "monitor": {"name": str, "status": str, "monitoringActive": bool},
    "pingStatus": "up" | "down",
}

opshift.alert(webhook_url, *, title, ...)

Trigger a webhook alert.

python
res = opshift.alert(
    "deploy-failures",
    title="Deploy failed: api-gateway",
    severity="sev1",
    description="Migration step exited with code 137",
    metadata={"commit": "a1b2c3", "environment": "production"},
    group_key="deploy:api-gateway",
)
 
if res["status"] == "filtered":
    print("Suppressed:", res.get("reason"))

Keyword arguments

FieldTypeDescription
titlerequiredstrAlert title, max 200 chars.
descriptionstrLong-form context, max 2000 chars.
severity"sev0".."sev4"Overrides the webhook default.
metadatadictMax 100 KB, max 10 nesting levels.
group_keystrMax 200 chars.

Return value

python
{
    "alertId": str | None,
    "status": "created" | "filtered" | "occurrence_added" | "reopened",
    "message": str,
    "occurrenceCount": int | None,
    "reason": str | None,
}

Error handling

All failures raise OpShiftError. Network-level failures (DNS, connection refused, timeout) surface with status_code == 0.

python
from opshift import OpShift, OpShiftError
 
try:
    opshift.ping("monitor-id")
except OpShiftError as error:
    print(error.status_code)  # 0, 401, 404, 429, ...
    print(str(error))         # human-readable
    print(error.body)         # parsed JSON body if available
    raise

Async usage

The client is synchronous. To call it from async code, dispatch to a thread:

python
import asyncio
from opshift import OpShift
 
opshift = OpShift(secret=...)
 
async def heartbeat():
    await asyncio.to_thread(opshift.ping, "your-ping-token")