base

dev_tool.services.runner.base

log = logging.getLogger(__name__) module-attribute

BaseProcessRunner

Bases: ABC

An abstract base class for process runners.

This class handles the common process lifecycle including starting, streaming output, and stopping a subprocess.

The constructor for the BaseProcessRunner class.

Parameters:

• project_name (str) –

  The project name for container naming.

Source code in dev_tool/services/runner/base.py

def __init__(self, project_name: str) -> None:
    """
    The constructor for the BaseProcessRunner class.

    :param project_name: The project name for container naming.
    """

    self.environment: dict[str, str] = {}
    self.interrupt_time = 0.0
    self.output_thread: threading.Thread | None = None
    self.process: subprocess.Popen | None = None
    self.project_name = project_name
    self.ready_event = threading.Event()
    self.stop_event = threading.Event()

environment = {} instance-attribute

interrupt_time = 0.0 instance-attribute

output_thread = None instance-attribute

process = None instance-attribute

project_name = project_name instance-attribute

ready_event = threading.Event() instance-attribute

stop_event = threading.Event() instance-attribute

force_kill

A method that forcefully kills the subprocess without waiting for cleanup.

Source code in dev_tool/services/runner/base.py

def force_kill(self) -> None:
    """A method that forcefully kills the subprocess without waiting for cleanup."""

    self.stop_event.set()
    self._kill_process()

is_running

A method that checks if the subprocess is alive.

Returns:

• bool –

  True if the subprocess is running, False otherwise.

Source code in dev_tool/services/runner/base.py

def is_running(self) -> bool:
    """
    A method that checks if the subprocess is alive.

    :return: True if the subprocess is running, False otherwise.
    """

    return self.process is not None and self.process.poll() is None

restart

A method that stops the subprocess and starts it again.

Source code in dev_tool/services/runner/base.py

def restart(self) -> None:
    """A method that stops the subprocess and starts it again."""

    self.stop()

    self.stop_event.clear()
    self.ready_event.clear()

    self.run()

run

A method that starts the subprocess and output thread.

Source code in dev_tool/services/runner/base.py

def run(self) -> None:
    """A method that starts the subprocess and output thread."""

    if not self._is_available():
        return

    if self._is_containerized() and not self._container_exists():
        return

    if not self._is_containerized():
        self._stop_container()

    creationflags = 0
    preexec_fn = None

    if sys.platform == OperatingSystem.WINDOWS:
        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
    else:
        preexec_fn = child_preexec

    command = [str(arg) for arg in self._build_command()]

    env = None

    if self.environment:
        env = {**os.environ, **self.environment}

    self.process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        creationflags=creationflags,
        preexec_fn=preexec_fn,  # noqa: PLW1509
        env=env
    )

    self.output_thread = threading.Thread(target=self._stream_output, daemon=True)
    self.output_thread.start()

run_and_wait

A method that starts the subprocess and blocks until it exits or is interrupted.

Source code in dev_tool/services/runner/base.py

def run_and_wait(self) -> None:
    """A method that starts the subprocess and blocks until it exits or is interrupted."""

    self.run()

    if self.process is None:
        return

    # The handler must never raise: an async KeyboardInterrupt can land
    # mid-cleanup and abort it, orphaning the child tree. Repeat
    # interrupts escalate to a group kill instead, debounced because
    # frozen builds can deliver one keypress as two signals.
    def handle_interrupt(_signal: int, _frame: object) -> None:
        now = time.monotonic()

        if self.stop_event.is_set():
            if now - self.interrupt_time > INTERRUPT_ESCALATION_SECONDS:
                self._kill_process()

            return

        self.interrupt_time = now
        self.stop_event.set()

    handler = signal.getsignal(signal.SIGINT)

    try:
        signal.signal(signal.SIGINT, handle_interrupt)
        signal_registered = True
    except ValueError:
        signal_registered = False

    try:
        wait_for_stop(Terminal(), self.stop_event, restart=self.restart, is_running=self.is_running)
    except KeyboardInterrupt:
        self.stop_event.set()
    finally:
        self.stop()

        if signal_registered:
            signal.signal(signal.SIGINT, handler)

stop

A method that stops the subprocess and cleans up resources.

Source code in dev_tool/services/runner/base.py

def stop(self) -> None:
    """A method that stops the subprocess and cleans up resources."""

    self.stop_event.set()

    if self._is_containerized():
        self._stop_container()

    self._stop_process()

    if self.process:
        try:
            self.process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self._kill_process()

            with contextlib.suppress(subprocess.TimeoutExpired):
                self.process.wait(timeout=5)

        # The direct child may exit while descendants that ignore SIGINT
        # survive in its process group, keeping the stdout pipe open, so
        # sweep the whole group once the child is gone.
        if sys.platform != OperatingSystem.WINDOWS:
            with contextlib.suppress(ProcessLookupError):
                os.killpg(self.process.pid, signal.SIGKILL)

        if self.output_thread and self.output_thread.is_alive():
            self.output_thread.join(timeout=2)

        # Closing stdout while the output thread is blocked in readline()
        # deadlocks on the buffered reader lock, so only close once the
        # thread has exited.
        if self.process.stdout and not (self.output_thread and self.output_thread.is_alive()):
            self.process.stdout.close()

wait_until_ready

A method that blocks until the process signals readiness or the timeout expires.

Parameters:

• timeout (float, default: 30.0 ) –

  The maximum time to wait in seconds.

Returns:

• bool –

  True if the process is ready, False if the timeout was reached.

Source code in dev_tool/services/runner/base.py

def wait_until_ready(self, timeout: float = 30.0) -> bool:
    """
    A method that blocks until the process signals readiness or the timeout expires.

    :param timeout: The maximum time to wait in seconds.
    :return: True if the process is ready, False if the timeout was reached.
    """

    return self.ready_event.wait(timeout=timeout)