"""Wait for app liveness and readiness from a CI or Compose network.""" from __future__ import annotations import argparse import time import urllib.request from urllib.parse import urlsplit def wait_for_health(base_url: str, timeout: float) -> None: parsed = urlsplit(base_url) if parsed.scheme not in {"http", "https"} or parsed.hostname is None: raise ValueError("base URL must be an absolute HTTP(S) URL") urls = (f"{base_url.rstrip('/')}/health/live", f"{base_url.rstrip('/')}/health/ready") deadline = time.monotonic() + timeout last_error = "unknown health failure" while time.monotonic() < deadline: try: for url in urls: with urllib.request.urlopen(url, timeout=1) as response: # noqa: S310 if response.status != 200: raise OSError(f"{url} returned HTTP {response.status}") return except OSError as exc: last_error = str(exc) time.sleep(0.5) raise TimeoutError(f"Timed out waiting for app health: {last_error}") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--base-url", default="http://127.0.0.1:8080") parser.add_argument("--timeout", type=float, default=30) args = parser.parse_args() wait_for_health(args.base_url, args.timeout) print("App liveness and readiness ready") if __name__ == "__main__": main()