def wait_for_http()

in nuvolaris/util.py [0:0]

• 33 lines of code
• 10 McCabe index (conditional complexity)

def wait_for_http(url: str, timeout: int = 60, up_statuses: List[Union[int, str]] = [200]):
    """Wait until an HTTP endpoint becomes available with an accepted status code.

    Args:
        url (str): Full URL to check (e.g. http://milvus:9091/healthz)
        timeout (int): Total seconds to wait before giving up.
        up_statuses (List[Union[int, str]]): Status codes or patterns considered as 'UP'.

    Raises:
        TimeoutError: If the endpoint doesn't respond with a valid status within the timeout.
    """
    parsed = urlparse(url)
    scheme = parsed.scheme
    host = parsed.hostname
    port = parsed.port or (443 if scheme == "https" else 80)
    path = parsed.path or "/"

    if scheme == "https":
        conn = urllib3.connectionpool.HTTPSConnectionPool(host, port=port,
                                                          timeout=urllib3.util.Timeout(connect=5.0, read=5.0),
                                                          retries=False)
    else:
        conn = urllib3.connectionpool.HTTPConnectionPool(host, port=port,
                                                         timeout=urllib3.util.Timeout(connect=5.0, read=5.0),
                                                         retries=False)

    deadline = time.time() + timeout

    while time.time() < deadline:
        try:
            response = conn.request("GET", path)
            if status_matches(response.status, up_statuses):
                logging.info(f"Service is up: {url} (status {response.status})")
                return
            else:
                logging.warning(f"Service responded with {response.status}, not in {up_statuses}. Waiting...")
        except (NewConnectionError, MaxRetryError):
            logging.warning(f"Cannot connect to {url}, retrying...")
        except ProtocolError as e:
            if "Connection reset by peer" in str(e):
                logging.warning("Connection reset by peer. Sleeping 2 seconds...")
                time.sleep(2)
                continue
            else:
                logging.error(f"Protocol error: {e}")
        time.sleep(1)