1import os
  2import subprocess
  3import sys
  4import time
  5import tomllib
  6from pathlib import Path
  7from typing import Any
  8
  9import click
 10
 11from plain.runtime import APP_PATH, PLAIN_TEMP_PATH
 12
 13from .process import ProcessManager
 14from .utils import has_pyproject_toml
 15
 16
 17def auto_start_services() -> None:
 18    """Start dev *services* in the background if not already running."""
 19
 20    # Check if we're in CI and auto-start is not explicitly enabled
 21    if os.environ.get("CI") and os.environ.get("DEV_SERVICES_AUTO") is None:
 22        return
 23
 24    if os.environ.get("DEV_SERVICES_AUTO", "true") not in [
 25        "1",
 26        "true",
 27        "yes",
 28    ]:
 29        return
 30
 31    # Only auto-start services for commands that need the database/runtime
 32    service_commands = {
 33        "db",
 34        "dev",
 35        "makemigrations",
 36        "migrate",
 37        "migrations",
 38        "preflight",
 39        "request",
 40        "run",
 41        "shell",
 42        "test",
 43    }
 44    if not (service_commands & set(sys.argv)):
 45        return
 46
 47    # Don't do anything if it looks like a "services" command is being run explicitly
 48    if "dev" in sys.argv:
 49        if "logs" in sys.argv or "services" in sys.argv or "--stop" in sys.argv:
 50            return
 51
 52    if not ServicesProcess.get_services(APP_PATH.parent):
 53        return
 54
 55    if ServicesProcess.running_pid():
 56        return
 57
 58    click.secho(
 59        "Starting background dev services (terminate with `plain dev --stop`)...",
 60        dim=True,
 61    )
 62
 63    subprocess.Popen(
 64        [sys.executable, "-m", "plain", "dev", "services", "--start"],
 65        start_new_session=True,
 66        stdout=subprocess.DEVNULL,
 67        stderr=subprocess.DEVNULL,
 68    )
 69
 70    # Give services time to start and retry the check
 71    wait_times = [0.5, 1, 1]  # First check at 0.5s, then 1s intervals
 72    for wait_time in wait_times:
 73        time.sleep(wait_time)
 74        if ServicesProcess.running_pid():
 75            return  # Services started successfully
 76
 77    # Only show error after multiple attempts
 78    if not ServicesProcess.running_pid():
 79        click.secho(
 80            "Failed to start dev services. Here are the logs:",
 81            fg="red",
 82        )
 83        subprocess.run(
 84            ["plain", "dev", "logs", "--services"],
 85            check=False,
 86        )
 87        sys.exit(1)
 88
 89
 90class ServicesProcess(ProcessManager):
 91    pidfile = PLAIN_TEMP_PATH / "dev" / "services.pid"
 92    log_dir = PLAIN_TEMP_PATH / "dev" / "logs" / "services"
 93
 94    @staticmethod
 95    def get_services(root: str | Path) -> dict[str, Any]:
 96        if not has_pyproject_toml(root):
 97            return {}
 98
 99        with open(Path(root, "pyproject.toml"), "rb") as f:
100            pyproject = tomllib.load(f)
101
102        return (
103            pyproject.get("tool", {})
104            .get("plain", {})
105            .get("dev", {})
106            .get("services", {})
107        )
108
109    def run(self) -> None:
110        self.write_pidfile()
111        self.prepare_log()
112        self.init_poncho(print)
113
114        assert self.poncho is not None, "poncho should be initialized"
115
116        try:
117            services = self.get_services(APP_PATH.parent)
118            for name, data in services.items():
119                env = {
120                    **os.environ,
121                    "PYTHONUNBUFFERED": "true",
122                    "FORCE_COLOR": "1",
123                    **data.get("env", {}),
124                }
125                self.poncho.add_process(name, data["cmd"], env=env)
126
127            self.poncho.loop()
128        finally:
129            self.rm_pidfile()
130            self.close()