Process Management

class NgrokProcess:
    """
    Container for ngrok process information and operations.
    """
    proc: subprocess.Popen     # The child process running ngrok
    pyngrok_config: PyngrokConfig  # Configuration used with ngrok
    api_url: str              # API URL for ngrok web interface
    logs: list               # Recent logs from ngrok (limited by max_logs)
    startup_error: str       # Error message if startup failed

    def healthy(self):
        """
        Check if ngrok process is running and healthy.

        Returns:
        bool: True if process is started, running, and healthy

        Raises:
        PyngrokSecurityError: When API URL is not supported
        """

    def start_monitor_thread(self):
        """
        Start a thread to monitor the ngrok process and its logs.
        Does nothing if monitor thread is already running.
        """

    def stop_monitor_thread(self):
        """
        Stop the monitor thread after the next log event.
        Sets a flag for the thread to terminate when it wakes up.
        """

from pyngrok import ngrok

# Get the current ngrok process
process = ngrok.get_ngrok_process()

print(f"Process ID: {process.proc.pid}")
print(f"API URL: {process.api_url}")
print(f"Is healthy: {process.healthy()}")

# Access recent logs
for log in process.logs[-5:]:  # Last 5 logs
    print(f"{log.t} [{log.lvl}] {log.msg}")

# Control monitoring
process.start_monitor_thread()
process.stop_monitor_thread()

def get_process(pyngrok_config):
    """
    Get the current ngrok process for the given config's ngrok_path.
    If ngrok is not running, starts a new process.

    Parameters:
    - pyngrok_config (PyngrokConfig): Configuration for ngrok interaction

    Returns:
    NgrokProcess: The ngrok process instance
    """

from pyngrok import process, conf

# Get process with default configuration
default_config = conf.get_default()
ngrok_process = process.get_process(default_config)

# Get process with custom configuration
custom_config = conf.PyngrokConfig(
    startup_timeout=30,
    monitor_thread=True
)
custom_process = process.get_process(custom_config)

print(f"Process API URL: {custom_process.api_url}")

def get_ngrok_process(pyngrok_config=None):
    """
    Get the current ngrok process for the given configuration.
    Installs ngrok if not present and starts the process if not running.

    Parameters:
    - pyngrok_config (PyngrokConfig, optional): Configuration override

    Returns:
    NgrokProcess: The running ngrok process instance

    Raises:
    PyngrokError: When process startup fails
    """

from pyngrok import ngrok, conf

# Get process with default configuration (installs and starts if needed)
process = ngrok.get_ngrok_process()
print(f"Process API URL: {process.api_url}")
print(f"Process healthy: {process.healthy()}")

# Get process with custom configuration
custom_config = conf.PyngrokConfig(
    startup_timeout=30,
    monitor_thread=True
)
custom_process = ngrok.get_ngrok_process(custom_config)
print(f"Custom process logs: {len(custom_process.logs)}")

# Access process details
print(f"Process PID: {process.proc.pid}")
print(f"Config path: {process.pyngrok_config.config_path}")

def is_process_running(ngrok_path):
    """
    Check if ngrok process is currently running for the given path.

    Parameters:
    - ngrok_path (str): Path to the ngrok binary

    Returns:
    bool: True if ngrok is running from the given path
    """

from pyngrok import process, conf

# Check if default ngrok is running
default_config = conf.get_default()
is_running = process.is_process_running(default_config.ngrok_path)
print(f"ngrok running: {is_running}")

# Check custom path
custom_path = "/usr/local/bin/ngrok"
is_custom_running = process.is_process_running(custom_path)
print(f"Custom ngrok running: {is_custom_running}")

def kill_process(ngrok_path):
    """
    Terminate the ngrok process for the given path.
    This method issues a kill request and does not block.

    Parameters:
    - ngrok_path (str): Path to the ngrok binary
    """

from pyngrok import process, conf

# Kill default ngrok process
default_config = conf.get_default()
process.kill_process(default_config.ngrok_path)

# Kill custom ngrok process
process.kill_process("/usr/local/bin/ngrok")

# Verify process is stopped
is_running = process.is_process_running(default_config.ngrok_path)
print(f"Process still running: {is_running}")  # Should be False

def run_process(ngrok_path, args):
    """
    Start a blocking ngrok process with the given path and arguments.
    This method is for direct ngrok invocation and blocks until completion.

    Parameters:
    - ngrok_path (str): Path to the ngrok binary
    - args (list): Arguments to pass to ngrok

    Note: Not compatible with non-blocking API methods
    """

def capture_run_process(ngrok_path, args):
    """
    Run ngrok process and capture its output.
    Blocks until process completes and returns captured output.

    Parameters:
    - ngrok_path (str): Path to the ngrok binary
    - args (list): Arguments to pass to ngrok

    Returns:
    str: Captured output from the ngrok process

    Raises:
    PyngrokNgrokError: When ngrok exits with non-zero code
    CalledProcessError: When process execution fails
    """

from pyngrok import process, conf

config = conf.get_default()

# Run ngrok with custom arguments (blocking)
# This is equivalent to running "ngrok version" from command line
try:
    process.run_process(config.ngrok_path, ["version"])
except Exception as e:
    print(f"Error running ngrok: {e}")

# Capture output from ngrok command
try:
    version_output = process.capture_run_process(config.ngrok_path, ["--version"])
    print(f"ngrok version: {version_output}")
except Exception as e:
    print(f"Error getting version: {e}")

# Run help command and capture output
help_output = process.capture_run_process(config.ngrok_path, ["--help"])
print(help_output)

def set_auth_token(pyngrok_config, token):
    """
    Set the ngrok auth token in the config file.

    Parameters:
    - pyngrok_config (PyngrokConfig): Configuration for ngrok interaction
    - token (str): The auth token to set

    Raises:
    PyngrokError: When ngrok_version is not supported
    PyngrokNgrokError: When ngrok could not start or save the token
    """

def set_api_key(pyngrok_config, key):
    """
    Set the ngrok API key in the config file (v3 only).

    Parameters:
    - pyngrok_config (PyngrokConfig): Configuration for ngrok interaction  
    - key (str): The API key to set

    Raises:
    PyngrokError: When ngrok_version is not supported
    PyngrokNgrokError: When ngrok could not start or save the key
    """

from pyngrok import process, conf

config = conf.get_default()

# Set auth token
try:
    process.set_auth_token(config, "your_authtoken_here")
    print("Auth token saved successfully")
except Exception as e:
    print(f"Error setting auth token: {e}")

# Set API key (ngrok v3 only)
if config.ngrok_version == "v3":
    try:
        process.set_api_key(config, "your_api_key_here")
        print("API key saved successfully")
    except Exception as e:
        print(f"Error setting API key: {e}")

from pyngrok import process, conf
import time

def log_callback(log):
    print(f"[{log.t}] {log.lvl}: {log.msg}")
    if log.err:
        print(f"  Error: {log.err}")

# Configuration with logging
config = conf.PyngrokConfig(
    log_event_callback=log_callback,
    monitor_thread=True,
    max_logs=50
)

# Get process with monitoring
ngrok_process = process.get_process(config)

# Check process health periodically
for i in range(5):
    time.sleep(2)
    if ngrok_process.healthy():
        print(f"Process healthy at check {i+1}")
    else:
        print(f"Process unhealthy at check {i+1}")
        if ngrok_process.startup_error:
            print(f"Startup error: {ngrok_process.startup_error}")

# Access accumulated logs
print(f"\nTotal logs captured: {len(ngrok_process.logs)}")
for log in ngrok_process.logs[-3:]:  # Last 3 logs
    print(f"  {log.lvl}: {log.msg}")

from pyngrok import process, conf

# Configuration for production environment
production_config = conf.PyngrokConfig(
    startup_timeout=60,        # Longer timeout for slow systems
    monitor_thread=True,       # Monitor for stability
    max_logs=500,             # Keep more logs
    request_timeout=15.0,     # Longer API timeouts
    start_new_session=True    # Isolate process on POSIX
)

# High-availability setup with health checking
def health_monitor(ngrok_process):
    """Monitor ngrok process health and restart if needed"""
    if not ngrok_process.healthy():
        print("Process unhealthy, restarting...")
        process.kill_process(ngrok_process.pyngrok_config.ngrok_path)
        time.sleep(1)
        return process.get_process(ngrok_process.pyngrok_config)
    return ngrok_process

# Start monitored process
monitored_process = process.get_process(production_config)

# Periodic health checks (in real application, use threading)
import time
for _ in range(10):
    monitored_process = health_monitor(monitored_process)
    time.sleep(5)