Run services automatically in pre-commit

Author: davegaeddert

bolt-dev/bolt/dev/precommit/cli.py

@@ -6,6 +6,8 @@
 
 from bolt.cli.print import print_event
 
+from ..services import Services
+
 try:
     import tomllib
 except ModuleNotFoundError:
@@ -38,60 +40,63 @@ def cli(install):
 
     pyproject = Path("pyproject.toml")
 
-    if pyproject.exists():
-        with open(pyproject, "rb") as f:
-            pyproject = tomllib.load(f)
-        for name, data in (
-            pyproject.get("tool", {})
-            .get("bolt", {})
-            .get("pre-commit", {})
-            .get("run", {})
-        ).items():
-            cmd = data["cmd"]
-            print_event(f"Custom: {name} -> {cmd}")
-            result = subprocess.run(cmd, shell=True)
-            if result.returncode != 0:
-                sys.exit(result.returncode)
-
-    # Run this first since it's probably the most likely to fail
-    if find_spec("bolt.code"):
-        check_short("Running bolt code checks", "bolt", "code", "check")
-
-    if Path("poetry.lock").exists():
-        check_short("Checking poetry.lock", "poetry", "lock", "--check")
-
-    if bolt_db_connected():
-        check_short(
-            "Running preflight checks",
-            "bolt",
-            "preflight",
-            "--database",
-            "default",
-        )
-        check_short("Checking Bolt migrations", "bolt", "legacy", "migrate", "--check")
-        check_short(
-            "Checking for Bolt models missing migrations",
-            "bolt",
-            "legacy",
-            "makemigrations",
-            "--dry-run",
-            "--check",
-        )
-    else:
-        check_short("Running Bolt checks (without database)", "bolt", "preflight")
-        click.secho("--> Skipping migration checks", bold=True, fg="yellow")
-
-    print_event("Running bolt compile")
-    result = subprocess.run(["bolt", "compile"])
-    if result.returncode != 0:
-        sys.exit(result.returncode)
+    with Services():
+        if pyproject.exists():
+            with open(pyproject, "rb") as f:
+                pyproject = tomllib.load(f)
+            for name, data in (
+                pyproject.get("tool", {})
+                .get("bolt", {})
+                .get("pre-commit", {})
+                .get("run", {})
+            ).items():
+                cmd = data["cmd"]
+                print_event(f"Custom: {name} -> {cmd}")
+                result = subprocess.run(cmd, shell=True)
+                if result.returncode != 0:
+                    sys.exit(result.returncode)
+
+        # Run this first since it's probably the most likely to fail
+        if find_spec("bolt.code"):
+            check_short("Running bolt code checks", "bolt", "code", "check")
+
+        if Path("poetry.lock").exists():
+            check_short("Checking poetry.lock", "poetry", "lock", "--check")
+
+        if bolt_db_connected():
+            check_short(
+                "Running preflight checks",
+                "bolt",
+                "preflight",
+                "--database",
+                "default",
+            )
+            check_short(
+                "Checking Bolt migrations", "bolt", "legacy", "migrate", "--check"
+            )
+            check_short(
+                "Checking for Bolt models missing migrations",
+                "bolt",
+                "legacy",
+                "makemigrations",
+                "--dry-run",
+                "--check",
+            )
+        else:
+            check_short("Running Bolt checks (without database)", "bolt", "preflight")
+            click.secho("--> Skipping migration checks", bold=True, fg="yellow")
 
-    if find_spec("bolt.pytest"):
-        print_event("Running tests")
-        result = subprocess.run(["bolt", "test"])
+        print_event("Running bolt compile")
+        result = subprocess.run(["bolt", "compile"])
         if result.returncode != 0:
             sys.exit(result.returncode)
 
+        if find_spec("bolt.pytest"):
+            print_event("Running tests")
+            result = subprocess.run(["bolt", "test"])
+            if result.returncode != 0:
+                sys.exit(result.returncode)
+
 
 def bolt_db_connected():
     result = subprocess.run(

bolt-dev/bolt/dev/services.py

@@ -1,6 +1,7 @@
 import os
 import subprocess
 import time
+from importlib.util import find_spec
 from pathlib import Path
 
 import click
@@ -46,15 +47,31 @@ def run(self):
         self.manager.loop()
 
     def __enter__(self):
+        if not self.get_services(APP_PATH.parent):
+            # No-op if no services are defined
+            return
+
         if Pid().exists():
             click.secho("Services already running in `bolt dev` command", fg="yellow")
             return
 
         print("Starting `bolt dev services`")
-        self.subprocess = subprocess.Popen(
+        self._subprocess = subprocess.Popen(
             ["bolt", "dev", "services"], cwd=APP_PATH.parent
         )
-        time.sleep(0.5)
+
+        if find_spec("bolt.db"):
+            time.sleep(0.5)  # Give it a chance to hit on the first try
+            subprocess.check_call(["bolt", "db", "wait"], env=os.environ)
+        else:
+            # A bit of a hack to wait for the services to start
+            time.sleep(3)
 
     def __exit__(self, *args):
-        self.subprocess.terminate()
+        if not hasattr(self, "_subprocess"):
+            return
+
+        self._subprocess.terminate()
+
+        # Flush the buffer so the output doesn't spill over
+        self._subprocess.communicate()