58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
"""Run the safe MDK WCS readiness probe through the canonical GeoIntel API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
|
|
import requests
|
|
|
|
|
|
DEFAULT_API_URL = "http://127.0.0.1:8000"
|
|
DEFAULT_PROJECT_NAME = "Kempen Regional Workbench"
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Probe MDK bathymetry WCS readiness without downloading coverage.")
|
|
parser.add_argument("--base-url", default=os.environ.get("GEOINTEL_INTERNAL_API_URL", DEFAULT_API_URL))
|
|
parser.add_argument("--project-name", default=DEFAULT_PROJECT_NAME)
|
|
parser.add_argument("--timeout", type=int, default=60)
|
|
return parser.parse_args()
|
|
|
|
|
|
def unwrap(response: requests.Response):
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
if not isinstance(payload, dict) or "data" not in payload:
|
|
raise RuntimeError(f"Non-canonical API response from {response.url}")
|
|
return payload["data"]
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
base_url = args.base_url.rstrip("/")
|
|
session = requests.Session()
|
|
projects = unwrap(
|
|
session.get(
|
|
f"{base_url}/api/v1/projects",
|
|
params={"name": args.project_name, "limit": 1},
|
|
timeout=args.timeout,
|
|
)
|
|
)["items"]
|
|
if not projects:
|
|
raise RuntimeError(f"Project {args.project_name!r} was not found")
|
|
result = unwrap(
|
|
session.get(
|
|
f"{base_url}/api/v1/projects/{projects[0]['id']}/datasets/"
|
|
"bathymetry/sources/mdk_bcp_bathymetry/readiness",
|
|
timeout=args.timeout,
|
|
)
|
|
)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0 if result["status"] == "reachable" else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|