#!/usr/bin/env python3 """Wait for PostgreSQL TCP and Qdrant HTTP during local development.""" from __future__ import annotations import argparse import socket import time import urllib.request def wait_tcp(host: str, port: int, deadline: float) -> None: while time.monotonic() < deadline: try: with socket.create_connection((host, port), timeout=1): return except OSError: time.sleep(0.5) raise TimeoutError(f"Timed out waiting for {host}:{port}") def wait_http(url: str, deadline: float) -> None: while time.monotonic() < deadline: try: with urllib.request.urlopen(url, timeout=1) as response: # noqa: S310 - caller supplies fixed local HTTP URL if response.status < 500: return except OSError: time.sleep(0.5) raise TimeoutError(f"Timed out waiting for {url}") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--timeout", type=float, default=60) args = parser.parse_args() deadline = time.monotonic() + args.timeout wait_tcp("127.0.0.1", 5432, deadline) wait_http("http://127.0.0.1:6333/healthz", deadline) print("Services ready") if __name__ == "__main__": main()