# Openstatus Python SDK

[![PyPI](https://img.shields.io/pypi/v/openstatus.svg)](https://pypi.org/project/openstatus/)
[![Python versions](https://img.shields.io/pypi/pyversions/openstatus.svg)](https://pypi.org/project/openstatus/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

Official Python SDK for [Openstatus](https://www.openstatus.dev) — the
open-source status page and uptime monitoring platform.

> Status: pre-alpha. APIs may change without notice until v1.0.

## Table of contents

- [Features](#features)
- [Installation](#installation)
- [Quick start](#quick-start)
- [Authentication](#authentication)
- [Sync and async](#sync-and-async)
- [Custom HTTP client](#custom-http-client)
- [Services](#services)
  - [Health](#health)
  - [Monitor](#monitor)
  - [Status report](#status-report)
  - [Status page](#status-page)
  - [Maintenance](#maintenance)
  - [Notification](#notification)
- [Reference](#reference)
  - [Regions](#regions)
  - [Enums](#enums)
- [Error handling](#error-handling)
- [Recipes](#recipes)
- [Migration from the Node SDK](#migration-from-the-node-sdk)
- [Development](#development)
- [License](#license)

## Features

- **Sync and async clients** — `OpenstatusClient` and `AsyncOpenstatusClient`,
  both backed by `httpx`.
- **Typed messages** — request and response types are generated from the
  upstream protobuf schema, with `.pyi` stubs for editor and pyright support.
- **All services covered** — Health, Monitor, StatusReport, StatusPage,
  Maintenance, and Notification.
- **JSON over HTTP** — uses Connect-RPC's JSON mode; no binary protobuf
  required at runtime.
- **Predictable errors** — Connect-style error envelopes are mapped to a typed
  exception hierarchy.

## Installation

```bash
pip install openstatus
# or
uv add openstatus
```

Requires Python 3.10 or newer.

## Quick start

```python
from openstatus import OpenstatusClient
from openstatus._gen.openstatus.monitor.v1.assertions_pb2 import (
    NumberComparator,
    StatusCodeAssertion,
)
from openstatus._gen.openstatus.monitor.v1.http_monitor_pb2 import (
    HTTPMethod,
    HTTPMonitor,
)
from openstatus._gen.openstatus.monitor.v1.monitor_pb2 import Periodicity, Region
from openstatus._gen.openstatus.monitor.v1.service_pb2 import (
    CreateHTTPMonitorRequest,
    ListMonitorsRequest,
)

with OpenstatusClient() as client:
    created = client.monitor.v1.MonitorService.create_http_monitor(
        CreateHTTPMonitorRequest(
            monitor=HTTPMonitor(
                name="My API",
                url="https://api.example.com/health",
                method=HTTPMethod.HTTP_METHOD_GET,
                periodicity=Periodicity.PERIODICITY_1M,
                regions=[Region.REGION_FLY_AMS, Region.REGION_FLY_IAD],
                active=True,
                status_code_assertions=[
                    StatusCodeAssertion(
                        comparator=NumberComparator.NUMBER_COMPARATOR_EQUAL,
                        target=200,
                    )
                ],
            )
        )
    )
    print(f"Created monitor id={created.monitor.id}")

    monitors = client.monitor.v1.MonitorService.list_monitors(ListMonitorsRequest())
    print(f"Total monitors: {monitors.total_size}")
```

## Authentication

The SDK reads `OPENSTATUS_API_KEY` from the environment by default:

```bash
export OPENSTATUS_API_KEY="..."
```

Or pass it explicitly:

```python
from openstatus import ClientOptions, OpenstatusClient

client = OpenstatusClient(ClientOptions(api_key="..."))
```

Override the base URL with `OPENSTATUS_API_URL` or `ClientOptions(base_url=...)`.
The default is `https://api.openstatus.dev`.

## Sync and async

Both clients expose the same nested namespace path
(`client.<service>.v1.<ServiceName>.<method>`):

```python
from openstatus import OpenstatusClient
from openstatus._gen.openstatus.health.v1.health_pb2 import CheckRequest

with OpenstatusClient() as client:
    health = client.health.v1.HealthService.check(CheckRequest())
```

```python
import asyncio
from openstatus import AsyncOpenstatusClient
from openstatus._gen.openstatus.health.v1.health_pb2 import CheckRequest

async def main() -> None:
    async with AsyncOpenstatusClient() as client:
        health = await client.health.v1.HealthService.check(CheckRequest())
        print(health.status)

asyncio.run(main())
```

## Custom HTTP client

Supply your own `httpx.Client` to control timeouts, transports, proxies, or
TLS configuration:

```python
import httpx
from openstatus import ClientOptions, OpenstatusClient

http = httpx.Client(
    timeout=httpx.Timeout(connect=2.0, read=10.0, write=10.0, pool=10.0),
    transport=httpx.HTTPTransport(retries=3),
)
client = OpenstatusClient(ClientOptions(http_client=http))
```

When you pass a client in, the SDK does **not** close it; manage its lifetime
yourself.

## Services

Each method takes a typed protobuf request and returns a typed protobuf
response. Pass per-call headers via the keyword-only `headers=` argument.

### Health

```python
from openstatus._gen.openstatus.health.v1.health_pb2 import CheckRequest

health = client.health.v1.HealthService.check(CheckRequest())
```

### Monitor

```python
from openstatus._gen.openstatus.monitor.v1.service_pb2 import (
    GetMonitorRequest,
    ListMonitorsRequest,
    TriggerMonitorRequest,
)

client.monitor.v1.MonitorService.list_monitors(ListMonitorsRequest())
client.monitor.v1.MonitorService.get_monitor(GetMonitorRequest(id="abc"))
client.monitor.v1.MonitorService.trigger_monitor(TriggerMonitorRequest(id="abc"))
```

### Status report

```python
from openstatus._gen.openstatus.status_report.v1.service_pb2 import ListStatusReportsRequest

client.status_report.v1.StatusReportService.list_status_reports(ListStatusReportsRequest())
```

### Status page

```python
from openstatus._gen.openstatus.status_page.v1.service_pb2 import ListStatusPagesRequest

client.status_page.v1.StatusPageService.list_status_pages(ListStatusPagesRequest())
```

### Maintenance

```python
from openstatus._gen.openstatus.maintenance.v1.service_pb2 import ListMaintenancesRequest

client.maintenance.v1.MaintenanceService.list_maintenances(ListMaintenancesRequest())
```

### Notification

```python
from openstatus._gen.openstatus.notification.v1.service_pb2 import ListNotificationsRequest

client.notification.v1.NotificationService.list_notifications(ListNotificationsRequest())
```

## Reference

### Regions

Region constants live in `openstatus._gen.openstatus.monitor.v1.monitor_pb2`
as `Region`. Use the descriptor API to enumerate them:

```python
from openstatus._gen.openstatus.monitor.v1.monitor_pb2 import Region

print(list(Region.keys()))  # ["REGION_UNSPECIFIED", "REGION_FLY_AMS", ...]
print(Region.Name(1))       # "REGION_FLY_AMS"
print(Region.Value("REGION_FLY_AMS"))
```

### Enums

Common enums:

- `HTTPMethod` (`openstatus.monitor.v1.http_monitor_pb2`) — request method for HTTP monitors.
- `Periodicity` (`openstatus.monitor.v1.monitor_pb2`) — check frequency.
- `MonitorStatus`, `Region` (`openstatus.monitor.v1.monitor_pb2`).
- `NumberComparator`, `StringComparator`, `RecordComparator`
  (`openstatus.monitor.v1.assertions_pb2`) — assertion operators.
- `HTTPResponseLogRequestStatus`, `HTTPResponseLogTrigger`, `TimeRange`
  (`openstatus.monitor.v1.service_pb2`).

## Error handling

All transport errors derive from `OpenstatusError`:

```python
from openstatus import (
    AuthenticationError,
    NotFoundError,
    OpenstatusClient,
    OpenstatusError,
)
from openstatus._gen.openstatus.monitor.v1.service_pb2 import GetMonitorRequest

with OpenstatusClient() as client:
    try:
        client.monitor.v1.MonitorService.get_monitor(GetMonitorRequest(id="missing"))
    except NotFoundError as err:
        print("monitor not found:", err.connect_code, err.http_status)
    except AuthenticationError:
        print("check OPENSTATUS_API_KEY")
    except OpenstatusError as err:
        print("openstatus failed:", err.connect_code, err.details)
```

Each exception carries `connect_code`, `http_status`, `details`, and
`raw_body` attributes for inspection.

| Connect code         | Exception                |
| -------------------- | ------------------------ |
| `unauthenticated`    | `AuthenticationError`    |
| `not_found`          | `NotFoundError`          |
| `invalid_argument`   | `InvalidArgumentError`   |
| `permission_denied`  | `PermissionDeniedError`  |
| `resource_exhausted` | `RateLimitError`         |
| `unavailable`        | `ServiceUnavailableError`|
| anything else        | `OpenstatusError`        |

When the server response is not a Connect error envelope (e.g. a 502 HTML
body from a proxy), `ServiceUnavailableError` is raised with the raw body
preserved on `.raw_body`.

## Recipes

### FastAPI

```python
from fastapi import Depends, FastAPI
from openstatus import OpenstatusClient

app = FastAPI()
_client = OpenstatusClient()

def get_client() -> OpenstatusClient:
    return _client

@app.on_event("shutdown")
def shutdown() -> None:
    _client.close()

@app.get("/monitors")
def monitors(client: OpenstatusClient = Depends(get_client)):
    from openstatus._gen.openstatus.monitor.v1.service_pb2 import ListMonitorsRequest
    res = client.monitor.v1.MonitorService.list_monitors(ListMonitorsRequest())
    return {"total": res.total_size}
```

### Django

```python
# myapp/openstatus.py
from django.conf import settings
from openstatus import ClientOptions, OpenstatusClient

_singleton: OpenstatusClient | None = None

def client() -> OpenstatusClient:
    global _singleton
    if _singleton is None:
        _singleton = OpenstatusClient(ClientOptions(api_key=settings.OPENSTATUS_API_KEY))
    return _singleton
```

## Migration from the Node SDK

- Method names: `MonitorService.listMonitors(...)` → `MonitorService.list_monitors(...)`.
- Message field access: `monitor.fooBar` → `monitor.foo_bar`.
- Enum lookups: `Region[value]` (Node) → `Region.Name(value)` (Python).
- Headers per call: pass as keyword argument `headers={...}`.
- Authentication: same `OPENSTATUS_API_KEY` env var and `ClientOptions(api_key=...)`
  parameter.

## Development

```bash
uv sync
uv run pytest tests/unit
uv run pyright
uv run ruff check .
```

Integration tests against the live API:

```bash
OPENSTATUS_API_KEY=... uv run pytest tests/integration
```

Regenerating the protobuf message classes:

```bash
bash scripts/regen.sh
```

See [CONTRIBUTING.md](./CONTRIBUTING.md) for details.

For the design rationale (why JSON over HTTP, why a pinned Buf archive,
why sync+async from day one), see [docs/decisions.md](./docs/decisions.md).

## License

MIT — see [LICENSE](./LICENSE).