Cookbook: Pattern: Hot Reload (Dev Server)¶

# app/services.py
from pico_ioc import component

@component
class MyService:
    # --- EDIT THIS LINE WHILE THE SERVER IS RUNNING ---
    GREETING = "Hello, Version 1!"
    # --------------------------------------------------

    def __init__(self):
        print(f"MyService instance CREATED (Version: {self.GREETING})")

    def greet(self) -> str:
        return self.GREETING

# dev_server.py
import sys
import threading
import importlib
from flask import Flask
from pico_ioc import init, PicoContainer
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, FileSystemEvent

# --- 1. Global Container Setup ---

# We need a lock to safely swap the container
# from another thread (the watchdog thread).
_container_lock = threading.Lock()

# Optional: Debounce reloads to avoid duplicate triggers from editor save cycles
_reload_debounce_lock = threading.Lock()
_reload_debounce_timer: threading.Timer | None = None
_DEBOUNCE_SECONDS = 0.25

# The global container variable that all threads will share.
# We initialize it once at the start.
container: PicoContainer = init(modules=["app.services"])


def get_current_container() -> PicoContainer:
    """Safely get the current container."""
    with _container_lock:
        return container

def _schedule_reload():
    """Debounce reloads to avoid firing multiple times for a single save."""
    global _reload_debounce_timer
    with _reload_debounce_lock:
        if _reload_debounce_timer is not None:
            _reload_debounce_timer.cancel()
        _reload_debounce_timer = threading.Timer(_DEBOUNCE_SECONDS, reload_container)
        _reload_debounce_timer.daemon = True
        _reload_debounce_timer.start()

def reload_container():
    """
    The core hot-reload logic.
    This is called by the file watcher (debounced).
    """
    global container
    print("\n--- [RELOAD] Code change detected! ---")

    # Invalidate import caches so Python sees new files
    importlib.invalidate_caches()

    # Clear Python's import cache for our app package and submodules
    # to force a fresh import.
    for module_name in list(sys.modules.keys()):
        if module_name == "app" or module_name.startswith("app."):
            print(f"[RELOAD] Clearing module cache for: {module_name}")
            del sys.modules[module_name]

    try:
        # Build the new container *before* shutting
        # down the old one, to ensure a valid config.
        new_container = init(modules=["app.services"])

        # Safely swap the containers
        with _container_lock:
            old_container = container
            container = new_container

        print("[RELOAD] New container is active.")

        # Shut down the old one *after* the swap
        old_container.shutdown()
        print("[RELOAD] Old container shut down.")

    except Exception as e:
        print(f"\n[RELOAD] FAILED to reload container: {e}\n")
        # Keep the old container active if reload fails

    print("--- [RELOAD] Ready ---\n")

# --- 2. File Watcher (watchdog) Setup ---

class ReloadHandler(FileSystemEventHandler):
    """Watches for .py file changes and schedules a debounced reload."""

    @staticmethod
    def _is_relevant(event: FileSystemEvent) -> bool:
        # Ignore directories and __pycache__ or hidden files
        if event.is_directory:
            return False
        path = event.src_path
        if "__pycache__" in path:
            return False
        if not path.endswith(".py"):
            return False
        return True

    def on_modified(self, event: FileSystemEvent):
        if self._is_relevant(event):
            _schedule_reload()

    def on_created(self, event: FileSystemEvent):
        if self._is_relevant(event):
            _schedule_reload()

    def on_moved(self, event: FileSystemEvent):
        if self._is_relevant(event):
            _schedule_reload()

def start_file_watcher():
    """Starts the watchdog observer in a background thread."""
    path = "./app"  # Watch the 'app' directory
    event_handler = ReloadHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    print(f"--- Watchdog monitoring directory: {path} ---")
    return observer

# --- 3. Flask Server Setup ---

app = Flask(__name__)

@app.route("/")
def home():
    """
    This Flask view gets the container on every request.
    It will automatically get the new one after a reload.
    """
    current_container = get_current_container()

    # Get the service (it will be created and cached
    # by the current container instance).
    # If your pico-ioc version supports type-based lookups, you can also:
    #   from app.services import MyService
    #   service = current_container.get(MyService)
    service = current_container.get("app.services.MyService")

    return f"<h1>{service.greet()}</h1>"

# --- 4. Main Entry Point ---

if __name__ == "__main__":
    print("--- Starting dev server with hot-reload ---")

    # Start the file watcher in the background
    watcher = start_file_watcher()

    # Start the Flask app in the foreground
    # (use_reloader=False is crucial!)
    try:
        app.run(port=5000, debug=True, use_reloader=False)
    finally:
        watcher.stop()
        watcher.join()
        print("--- Server and watcher stopped ---")