Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
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())
|