Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -23,13 +23,15 @@ classifiers = [
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: Internet :: WWW/HTTP",
"Typing :: Typed",
"Environment :: Console",
]
dependencies = [
"httpx>=0.27,<1.0",
"pydantic>=2.0,<3.0",
]

[project.optional-dependencies]
cli = ["typer[all]>=0.12,<1.0"]
dev = [
"pytest>=9.0.3",
"pytest-cov",
Expand All @@ -38,8 +40,12 @@ dev = [
"mypy>=1.10",
"pydantic",
"pip-audit>=2.7",
"typer[all]>=0.12,<1.0",
]

[project.scripts]
hyp = "hyperping.cli._app:app"

[project.urls]
Homepage = "https://github.com/develeap/hyperping-python"
Documentation = "https://github.com/develeap/hyperping-python#readme"
Expand Down
Empty file added src/hyperping/cli/__init__.py
Empty file.
52 changes: 52 additions & 0 deletions src/hyperping/cli/_app.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
"""Main hyp CLI application."""

from __future__ import annotations

from typing import Annotated

import typer

from hyperping._version import __version__
from hyperping.cli._incidents import incident_app
from hyperping.cli._monitors import monitor_app
from hyperping.cli._statuspages import statuspage_app
from hyperping.cli._tenant import tenant_app

app = typer.Typer(
name="hyp",
help="Hyperping CLI: manage monitors, incidents, and status pages.",
no_args_is_help=True,
)

app.add_typer(monitor_app, name="monitor")
app.add_typer(incident_app, name="incident")
app.add_typer(statuspage_app, name="statuspage")
app.add_typer(tenant_app, name="tenant")


def _version_callback(value: bool) -> None:
if value:
typer.echo(f"hyp {__version__}")
raise typer.Exit()


@app.callback()
def main(
ctx: typer.Context,
api_key: Annotated[
str | None,
typer.Option("--api-key", envvar="HYPERPING_API_KEY", help="Hyperping API key."),
] = None,
json_output: Annotated[
bool,
typer.Option("--json", help="Output as JSON."),
] = False,
version: Annotated[
bool | None,
typer.Option("--version", callback=_version_callback, is_eager=True, help="Show version."),
] = None,
) -> None:
"""Hyperping CLI."""
ctx.ensure_object(dict)
ctx.obj["api_key"] = api_key
ctx.obj["json"] = json_output
24 changes: 24 additions & 0 deletions src/hyperping/cli/_config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
"""CLI client factory and API key resolution."""

from __future__ import annotations

import os

import typer

from hyperping.client import HyperpingClient


def get_client(api_key: str | None) -> HyperpingClient:
"""Resolve API key and return a configured client.

Resolution order: explicit flag value, then HYPERPING_API_KEY env var.
Raises typer.BadParameter if no key is available.
"""
key = api_key or os.environ.get("HYPERPING_API_KEY")
if not key:
raise typer.BadParameter(
"API key required. Pass --api-key or set HYPERPING_API_KEY.",
param_hint="--api-key",
)
return HyperpingClient(api_key=key)
87 changes: 87 additions & 0 deletions src/hyperping/cli/_incidents.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
"""hyp incident subcommands."""

from __future__ import annotations

from typing import Annotated

import typer

from hyperping.cli._config import get_client
from hyperping.cli._output import print_detail, print_error, print_success, print_table
from hyperping.exceptions import HyperpingAPIError
from hyperping.models import Incident, IncidentCreate, IncidentType, LocalizedText

incident_app = typer.Typer(name="incident", help="Manage incidents.")


def _incident_row(i: Incident) -> list[object]:
return [i.uuid, i.title_en, i.type, str(i.is_resolved), i.date or ""]


@incident_app.command("list")
def incident_list(
ctx: typer.Context,
status: Annotated[str | None, typer.Option("--status", help="Filter by status")] = None,
) -> None:
"""List incidents."""
api_key: str | None = ctx.obj.get("api_key")
json_mode: bool = ctx.obj.get("json", False)
client = get_client(api_key)
incidents = client.list_incidents(status=status)
columns = ["uuid", "title", "type", "resolved", "date"]
rows = [_incident_row(i) for i in incidents]
print_table("Incidents", columns, rows, json_mode)


@incident_app.command("create")
def incident_create(
ctx: typer.Context,
title: Annotated[str, typer.Option("--title", help="Incident title (English)")],
text: Annotated[str, typer.Option("--text", help="Incident message (English)")],
statuspage: Annotated[str, typer.Option("--statuspage", help="Status page UUID")],
incident_type: Annotated[
str | None, typer.Option("--type", help="Incident type: incident or outage")
] = None,
) -> None:
"""Create a new incident."""
api_key: str | None = ctx.obj.get("api_key")
json_mode: bool = ctx.obj.get("json", False)
client = get_client(api_key)
payload = IncidentCreate(
title=LocalizedText.from_string(title),
text=LocalizedText.from_string(text),
type=IncidentType(incident_type) if incident_type else IncidentType.INCIDENT,
statuspages=[statuspage],
)
try:
incident = client.create_incident(payload)
except HyperpingAPIError as exc:
print_error(str(exc))
raise typer.Exit(code=1) from exc
fields = {
"uuid": incident.uuid,
"title": incident.title_en,
"type": incident.type,
"resolved": incident.is_resolved,
"date": incident.date or "",
}
print_detail("Incident", fields, json_mode)


@incident_app.command("resolve")
def incident_resolve(
ctx: typer.Context,
incident_id: Annotated[str, typer.Argument(help="Incident UUID")],
message: Annotated[
str | None, typer.Option("--message", help="Resolution message")
] = None,
) -> None:
"""Resolve an incident."""
api_key: str | None = ctx.obj.get("api_key")
client = get_client(api_key)
try:
incident = client.resolve_incident(incident_id, message=message)
except HyperpingAPIError as exc:
print_error(str(exc))
raise typer.Exit(code=1) from exc
print_success(f"Incident {incident.uuid} resolved.")
90 changes: 90 additions & 0 deletions src/hyperping/cli/_monitors.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
"""hyp monitor subcommands."""

from __future__ import annotations

from typing import Annotated

import typer

from hyperping.cli._config import get_client
from hyperping.cli._output import print_detail, print_error, print_success, print_table
from hyperping.exceptions import HyperpingAPIError
from hyperping.models import Monitor

monitor_app = typer.Typer(name="monitor", help="Manage monitors.")


def _monitor_status(m: Monitor) -> str:
if m.paused:
return "paused"
return "down" if m.down else "up"


@monitor_app.command("list")
def monitor_list(ctx: typer.Context) -> None:
"""List all monitors."""
api_key: str | None = ctx.obj.get("api_key")
json_mode: bool = ctx.obj.get("json", False)
client = get_client(api_key)
monitors = client.list_monitors()
columns = ["uuid", "name", "url", "protocol", "status"]
rows = [[m.uuid, m.name, m.url, m.protocol, _monitor_status(m)] for m in monitors]
print_table("Monitors", columns, rows, json_mode)


@monitor_app.command("get")
def monitor_get(
ctx: typer.Context,
monitor_id: Annotated[str, typer.Argument(help="Monitor UUID")],
) -> None:
"""Show a single monitor."""
api_key: str | None = ctx.obj.get("api_key")
json_mode: bool = ctx.obj.get("json", False)
client = get_client(api_key)
try:
m = client.get_monitor(monitor_id)
except HyperpingAPIError as exc:
print_error(str(exc))
raise typer.Exit(code=1) from exc
fields = {
"uuid": m.uuid,
"name": m.name,
"url": m.url,
"protocol": m.protocol,
"status": _monitor_status(m),
"check_frequency": m.check_frequency,
"regions": ", ".join(m.regions),
}
print_detail("Monitor", fields, json_mode)


@monitor_app.command("pause")
def monitor_pause(
ctx: typer.Context,
monitor_id: Annotated[str, typer.Argument(help="Monitor UUID")],
) -> None:
"""Pause a monitor."""
api_key: str | None = ctx.obj.get("api_key")
client = get_client(api_key)
try:
m = client.pause_monitor(monitor_id)
except HyperpingAPIError as exc:
print_error(str(exc))
raise typer.Exit(code=1) from exc
print_success(f"Monitor {m.uuid} paused (status: {_monitor_status(m)})")


@monitor_app.command("resume")
def monitor_resume(
ctx: typer.Context,
monitor_id: Annotated[str, typer.Argument(help="Monitor UUID")],
) -> None:
"""Resume a paused monitor."""
api_key: str | None = ctx.obj.get("api_key")
client = get_client(api_key)
try:
m = client.resume_monitor(monitor_id)
except HyperpingAPIError as exc:
print_error(str(exc))
raise typer.Exit(code=1) from exc
print_success(f"Monitor {m.uuid} resumed (status: {_monitor_status(m)})")
58 changes: 58 additions & 0 deletions src/hyperping/cli/_output.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
"""CLI output formatters: rich tables, panels, and JSON."""

from __future__ import annotations

import json
from typing import Any

import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table

_console = Console()
_err_console = Console(stderr=True)


def print_table(
title: str,
columns: list[str],
rows: list[list[Any]],
json_mode: bool,
) -> None:
"""Render a list of rows as a rich table or a JSON array."""
if json_mode:
data = [dict(zip(columns, row)) for row in rows]
typer.echo(json.dumps(data, indent=2, default=str))
return

table = Table(title=title, show_header=True, header_style="bold")
for col in columns:
table.add_column(col)
for row in rows:
table.add_row(*[str(v) if v is not None else "" for v in row])
_console.print(table)


def print_detail(
title: str,
fields: dict[str, Any],
json_mode: bool,
) -> None:
"""Render a single record as a rich panel or a JSON object."""
if json_mode:
typer.echo(json.dumps(fields, indent=2, default=str))
return

lines = "\n".join(f"[bold]{k}[/bold]: {v}" for k, v in fields.items())
_console.print(Panel(lines, title=title))


def print_success(message: str) -> None:
"""Print a success message."""
_console.print(f"[green]{message}[/green]")


def print_error(message: str) -> None:
"""Print an error message to stderr."""
_err_console.print(f"[red]{message}[/red]")
Loading