Source code for harp.utils.network

import dataclasses
import socket
import time



[docs]
def get_available_network_port():
    import socket

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("", 0))
    port = sock.getsockname()[1]
    sock.close()
    return port




[docs]
def wait_for_port(port: int, host: str = "localhost", timeout: float = 10.0):
    """Wait until a port starts accepting TCP connections.
    Args:
        port: Port number.
        host: Host address on which the port should exist.
        timeout: In seconds. How long to wait before raising errors.
    Raises:
        TimeoutError: The port isn't accepting connection after time specified in `timeout`.
    """
    start_time = time.perf_counter()
    while True:
        try:
            with socket.create_connection((host, port), timeout=timeout):
                break
        except OSError as ex:
            time.sleep(0.01)
            if time.perf_counter() - start_time >= timeout:
                raise TimeoutError(
                    "Waited too long for the port {} on host {} to start accepting " "connections.".format(port, host)
                ) from ex




[docs]
@dataclasses.dataclass(frozen=True)
class Bind:
    host: str
    port: int

    def __str__(self):
        return f"{self.host}:{self.port}"