Tunnel 

Tunnel(
    port,
    check_local_port=True,
    debug=False,
    timeout=60,
    propagate=False,
    log_handlers=None,
    log_dir=None,
    callback=None,
)

Tunnel class for managing subprocess-based tunnels.

Parameters:
  • port (int) –

    The local port on which the tunnels will be created.

  • check_local_port (bool, default: True ) –

    Flag to check if the local port is available.

  • debug (bool, default: False ) –

    Flag to enable debug mode for additional output.

  • timeout (int, default: 60 ) –

    Maximum time to wait for the tunnels to start.

  • propagate (bool, default: False ) –

    Flag to propagate log messages to the root logger, if False will create custom log format to print log.

  • log_handlers (List[Handler], default: None ) –

    List of logging handlers to be added to the Tunnel logger.

  • log_dir (StrOrPath, default: None ) –

    Directory to store tunnel log files. If None it will set to os.get_cwd().

  • callback (Callable[[List[Tuple[str, Optional[str]]]], None], default: None ) –

    A callback function to be called when Tunnel URL is printed. will call callback([(url1, note1), (url2, note2), ...]) -> None.
Note

output of each tunnel command will be saved to log_dir

with_tunnel_list classmethod 

with_tunnel_list(
    port,
    tunnel_list,
    check_local_port=True,
    debug=False,
    timeout=60,
    propagate=False,
    log_handlers=None,
    log_dir=None,
    callback=None,
)

Create a Tunnel instance with a pre-defined list of tunnels.

Parameters:
  • port (int) –

    The local port on which the tunnels will be created.

  • tunnel_list (List[dict]) –

    List of dictionaries specifying tunnel configurations. Each dictionary must have the keys command, pattern, name, note (optional), and callback (optional).

  • check_local_port (bool, default: True ) –

    Flag to check if the local port is available.

  • debug (bool, default: False ) –

    Flag to enable debug mode for additional output.

  • timeout (int, default: 60 ) –

    Maximum time to wait for the tunnels to start.

  • propagate (bool, default: False ) –

    Flag to propagate log messages to the root logger, if False will create custom log format to print log.

  • log_handlers (List[Handler], default: None ) –

    List of logging handlers to be added to the Tunnel logger.

  • log_dir (StrOrPath, default: None ) –

    Directory to store tunnel log files. If None it will set to os.get_cwd().

  • callback (Callable[[List[Tuple[str, Optional[str], Optional[str]]]], None], default: None ) –

    A callback function to be called when Tunnel URL is printed. will call callback([(url1, note1, name1), (url2, note2, name2), ...]) -> None.
Raises:
  • ValueError –

    Raised if tunnel_list doesn't have dict with keys atleast command, pattern, name
Note

output of each tunnel command will be saved to log_dir

add_tunnel 

add_tunnel(
    *, command, pattern, name, note=None, callback=None
)

Add a tunnel.

Parameters:
  • command (str) –

    The command to execute for the tunnel.

  • pattern (StrOrRegexPattern) –

    A regular expression pattern to match the tunnel URL.

  • name (str) –

    The name of the tunnel.

  • note (str, default: None ) –

    A note about the tunnel. Defaults to None.

  • callback (Callable[[str, Optional[str], Optional[str]], None], default: None ) –

    A callback function to be called when when the regex pattern matched. will call callback(url, note, name) -> None. Defaults to None.
Note

name must be unique name as is being used for .log file,

start 

start()

Start the tunnel and wait for the URLs to be printed.

Raises:
  • RuntimeError –

    Raised if tunnel is already running

stop 

stop()

Stop the tunnel and reset internal state.

Raises:
  • RuntimeError –

    Raised if tunnel is not running

reset 

reset()

Reset internal state.

is_port_in_use staticmethod 

is_port_in_use(port)

Check if the specified port is in use.

Parameters:
  • port (int) –

    The port to check.
Returns:
  • bool( bool ) –

    True if the port is in use, False otherwise.

wait_for_condition staticmethod 

wait_for_condition(condition, *, interval=1, timeout=10)

Wait for the condition to be true until the specified timeout.

Mostly for internal use but can be used for anything else.

Parameters:
  • condition (Callable[[], bool]) –

    The condition to check.

  • interval (int, default: 1 ) –

    The interval (in seconds) between condition checks.

  • timeout (int, default: 10 ) –

    Maximum time to wait for the condition. None for no timeout.
Returns:
  • bool( bool ) –

    True if the condition is met, False if timeout is reached.