Sith/sith/composer.py

import logging
import signal
import subprocess
import sys
from pathlib import Path

import psutil

from sith import settings


def get_pid_file(procfile: Path) -> Path:
    """Get the PID file associated with a procfile"""
    return settings.BASE_DIR / procfile.with_suffix(f"{procfile.suffix}.pid")


def get_pid(procfile: Path) -> int | None:
    """Read the PID file to get the currently running composer if it exists"""
    file = get_pid_file(procfile)
    if not file.exists():
        return None
    with open(file, "r", encoding="utf8") as f:
        return int(f.read())


def write_pid(procfile: Path, pid: int):
    """Write currently running composer pid in PID file"""
    file = get_pid_file(procfile)
    if not file.exists():
        file.parent.mkdir(parents=True, exist_ok=True)
    with open(file, "w", encoding="utf8") as f:
        _ = f.write(str(pid))


def delete_pid(procfile: Path):
    """Delete PID file for cleanup"""
    get_pid_file(procfile).unlink(missing_ok=True)


def is_composer_running(procfile: Path) -> bool:
    """Check if the process in the PID file is running"""
    pid = get_pid(procfile)
    if pid is None:
        return False
    try:
        return psutil.Process(pid).is_running()
    except psutil.NoSuchProcess:
        return False


def start_composer(procfile: Path):
    """Starts the composer and stores the PID as an environment variable
    This allows for running smoothly with the django reloader
    """
    if is_composer_running(procfile):
        logging.info(
            f"Composer for {procfile} is already running with pid {get_pid(procfile)}"
        )
        logging.info(
            f"If this is a mistake, please delete {get_pid_file(procfile)} and restart the process"
        )
        return
    process = subprocess.Popen(
        [sys.executable, "-m", "honcho", "-f", str(procfile), "start"],
    )
    write_pid(procfile, process.pid)


def stop_composer(procfile: Path):
    """Stops the composer if it was started before"""
    if is_composer_running(procfile):
        process = psutil.Process(get_pid(procfile))
        if process.parent() != psutil.Process():
            logging.info(
                f"Currently running composer for {procfile} is controlled by another process"
            )
            return
        process.send_signal(signal.SIGTERM)
        _ = process.wait()
        delete_pid(procfile)