Initial upload: mockapi - OpenAPI Mock Server Generator

src/mockapi/core/hot_reload.py

@@ -0,0 +1,137 @@
+"""Hot-Reload functionality for spec file changes."""
+
+import os
+import subprocess
+import sys
+import threading
+import time
+from pathlib import Path
+from typing import Optional
+
+from watchdog.observers import Observer
+from watchdog.events import FileSystemEventHandler
+
+
+class HotReloadHandler(FileSystemEventHandler):
+    """Handler for file system events."""
+
+    def __init__(self, spec_file: str, debounce_ms: int = 500):
+        """Initialize the handler.
+
+        Args:
+            spec_file: Path to the spec file to watch
+            debounce_ms: Debounce delay in milliseconds
+        """
+        super().__init__()
+        self.spec_file = Path(spec_file).resolve()
+        self.debounce_ms = debounce_ms
+        self.last_reload_time = 0
+        self.reload_callback: Optional[callable] = None
+
+    def on_modified(self, event):
+        """Handle file modification events."""
+        if event.is_directory:
+            return
+
+        event_path = Path(event.src_path).resolve()
+
+        if event_path == self.spec_file:
+            current_time = time.time() * 1000
+            if current_time - self.last_reload_time > self.debounce_ms:
+                self.last_reload_time = current_time
+                self._trigger_reload()
+
+    def _trigger_reload(self):
+        """Trigger a reload."""
+        if self.reload_callback:
+            self.reload_callback()
+
+
+class HotReloader:
+    """Watches spec file and restarts server on changes."""
+
+    def __init__(
+        self,
+        spec_file: str,
+        port: int = 8080,
+        host: str = "0.0.0.0",
+        debounce_ms: int = 500,
+    ):
+        """Initialize the hot reloader.
+
+        Args:
+            spec_file: Path to the OpenAPI spec file
+            port: Server port
+            host: Server host
+            debounce_ms: Debounce delay for rapid changes
+        """
+        self.spec_file = Path(spec_file).resolve()
+        self.port = port
+        self.host = host
+        self.debounce_ms = debounce_ms
+
+        self.observer: Optional[Observer] = None
+        self.server_process: Optional[subprocess.Popen] = None
+        self._stop_event = threading.Event()
+
+    def start_watching(self):
+        """Start watching for file changes and run server."""
+        handler = HotReloadHandler(str(self.spec_file), self.debounce_ms)
+        handler.reload_callback = self._on_spec_changed
+
+        self.observer = Observer()
+        self.observer.schedule(
+            handler,
+            str(self.spec_file.parent),
+            recursive=False,
+        )
+        self.observer.start()
+
+        self._start_server()
+
+        try:
+            while not self._stop_event.is_set():
+                time.sleep(0.5)
+        except KeyboardInterrupt:
+            self.stop()
+
+    def stop(self):
+        """Stop watching and terminate server."""
+        self._stop_event.set()
+
+        if self.observer:
+            self.observer.stop()
+            self.observer.join()
+
+        if self.server_process:
+            self.server_process.terminate()
+            self.server_process.wait()
+
+    def _start_server(self):
+        """Start the mock server process."""
+        print(f"Starting mock server on {self.host}:{self.port}")
+
+        self.server_process = subprocess.Popen(
+            [
+                sys.executable,
+                "-m",
+                "uvicorn",
+                "mockapi.core.server_generator:create_mock_server",
+                "--host",
+                self.host,
+                "--port",
+                str(self.port),
+            ],
+            env={**os.environ, "MOCKAPI_SPEC": str(self.spec_file)},
+        )
+
+    def _on_spec_changed(self):
+        """Handle spec file changes."""
+        print(f"\nSpec file changed: {self.spec_file}")
+        print("Reloading server...")
+
+        if self.server_process:
+            self.server_process.terminate()
+            self.server_process.wait()
+
+        self._start_server()