874b4a5a02
- Set ruff/black line length to 120 - Reformatted code with black - Fixed import ordering with ruff - Disabled lint for UI component files with long CSS strings - Updated pyproject.toml with proper tool configuration
58 lines
1.9 KiB
Python
Executable File
58 lines
1.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Simple deployment health check utility."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import urlopen
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Poll an HTTP health endpoint until it is ready.")
|
|
parser.add_argument("url", help="Health endpoint URL, e.g. https://vault.example.com/health")
|
|
parser.add_argument("--timeout", type=int, default=120, help="Maximum wait time in seconds")
|
|
parser.add_argument("--interval", type=int, default=5, help="Poll interval in seconds")
|
|
parser.add_argument("--expect-status", default="ok", help="Expected JSON status field")
|
|
parser.add_argument(
|
|
"--expect-environment",
|
|
default=None,
|
|
help="Optional expected JSON environment field",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def fetch_json(url: str) -> dict:
|
|
with urlopen(url, timeout=10) as response:
|
|
body = response.read().decode("utf-8")
|
|
return json.loads(body)
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
deadline = time.time() + args.timeout
|
|
last_error = "unknown error"
|
|
|
|
while time.time() < deadline:
|
|
try:
|
|
payload = fetch_json(args.url)
|
|
if payload.get("status") != args.expect_status:
|
|
raise RuntimeError(f"unexpected status: {payload!r}")
|
|
if args.expect_environment and payload.get("environment") != args.expect_environment:
|
|
raise RuntimeError(f"unexpected environment: {payload!r}")
|
|
print(f"healthcheck passed: {payload}")
|
|
return 0
|
|
except (HTTPError, URLError, TimeoutError, ValueError, RuntimeError) as exc:
|
|
last_error = str(exc)
|
|
time.sleep(args.interval)
|
|
|
|
print(f"healthcheck failed after {args.timeout}s: {last_error}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|