Implement sanic-hmr: Hot Module Reloading wrapper CLI for Sanic

examples/sanic/app.py

@@ -0,0 +1,30 @@
+from sanic import Sanic, response
+
+
+def create_app():
+    # Use a unique name each time to avoid conflicts
+    import time
+
+    app_name = f"ExampleApp_{int(time.time() * 1000)}"
+    app = Sanic(app_name)
+
+    @app.route("/")
+    async def hello_world(_):
+        return response.text("Hello, world! (modified with factory)")
+
+    @app.route("/test")
+    async def test_route(_):
+        return response.json({"message": "This is a test route"})
+
+    @app.route("/api/users/<user_id:int>")
+    async def get_user(_, user_id):
+        return response.json({"user_id": user_id, "name": f"User {user_id}"})
+
+    return app
+
+
+# Export the app instance
+app = create_app()
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=8000, debug=True)

examples/sanic/pyproject.toml

@@ -0,0 +1,11 @@
+[project]
+name = "sanic-example"
+version = "0"
+requires-python = ">=3.12"
+dependencies = [
+    "sanic",
+    "sanic-hmr",
+]
+
+[tool.uv.sources]
+sanic-hmr = { workspace = true }

packages/sanic-hmr/README.md

@@ -0,0 +1,106 @@
+# sanic-hmr
+
+[![PyPI - Version](https://img.shields.io/pypi/v/sanic-hmr)](https://pypi.org/project/sanic-hmr/)
+
+This package provides hot module reloading (HMR) for [`sanic`](https://github.com/sanic-org/sanic).
+
+It uses [`watchfiles`](https://github.com/samuelcolvin/watchfiles) to detect FS modifications,
+re-executes the corresponding modules with [`hmr`](https://github.com/promplate/pyth-on-line/tree/main/packages/hmr) and restart the server (in the same process).
+
+**HOT** means the main process never restarts, and reloads are fine-grained (only the changed modules and their dependent modules are reloaded).
+Since the python module reloading is on-demand and the server is not restarted on every save, it is much faster than Sanic's built-in `--auto-reload` option.
+
+## Why?
+
+1. When you use `sanic app.py --auto-reload`, it restarts the whole process on every file change, but restarting the whole process is unnecessary:
+   - There is no need to restart the Python interpreter, neither all the 3rd-party packages you imported.
+   - Your changes usually affect only one single file, the rest of your application remains unchanged.
+2. `hmr` tracks dependencies at runtime, remembers the relationships between your modules and only reruns necessary modules.
+3. So you can save a lot of time by not restarting the whole process on every file change. You can see a significant speedup for debugging large applications.
+4. Although magic is involved, we thought and tested them very carefully, so everything works just as-wished.
+   - Your lazy loading through module-level `__getattr__` still works
+   - Your runtime imports through `importlib.import_module` or even `__import__` still work
+   - Even valid circular imports between `__init__.py` and sibling modules still work
+   - Fine-grained dependency tracking in the above cases still work
+   - Decorators still work, even meta programming hacks like `getsource` calls work too
+   - Standard dunder metadata like `__name__`, `__doc__`, `__file__`, `__package__` are correctly set
+
+Normally, you can replace `sanic app.py --auto-reload` with `sanic-hmr app:app` and everything will work as expected, with a much faster refresh experience.
+
+## Installation
+
+```sh
+pip install sanic-hmr
+```
+
+## Usage
+
+Replace
+
+```sh
+sanic app.py --auto-reload
+```
+
+or
+
+```sh
+python -m sanic app.app --auto-reload
+```
+
+with
+
+```sh
+sanic-hmr app:app
+```
+
+Everything will work as-expected, but with **hot** module reloading.
+
+## CLI Arguments
+
+This package supports the most common Sanic server configuration options:
+
+- `--host`: Host to listen on (default: `localhost`)
+- `--port`: Port to listen on (default: `8000`)
+- `--workers`: Number of worker processes (default: `1` - forced for HMR compatibility)
+- `--debug`: Enable debug mode (default: `False`)
+- `--access-log/--no-access-log`: Enable/disable access log (default: enabled)
+- `--clear`: Clear the terminal before each reload (default: `False`)
+
+The behavior of `reload_include` and `reload_exclude` is:
+
+1. Only file or directory paths are allowed; patterns will be treated as literal paths.
+2. Currently only supports hot-reloading Python source files.
+3. If you do not provide `reload_include`, the current directory is included by default; if you do provide it, only the specified paths are included. The same applies to `reload_exclude`.
+
+## Limitations
+
+- **Single worker only**: HMR requires running in single-worker mode. The `--workers` argument is ignored and forced to 1.
+- **Python files only**: Only Python source files are watched for changes.
+- **No WebSocket auto-reload**: Unlike some development servers, this doesn't automatically refresh browser pages.
+
+## Example
+
+Create a simple Sanic app:
+
+```python
+# app.py
+from sanic import Sanic, response
+
+app = Sanic("MyApp")
+
+@app.route("/")
+async def hello_world(request):
+    return response.text("Hello, world!")
+
+@app.route("/test")
+async def test_route(request):
+    return response.json({"message": "This is a test route"})
+```
+
+Then run with hot module reloading:
+
+```sh
+sanic-hmr app:app
+```
+
+Now when you modify `app.py`, the server will automatically reload only the changed modules, providing much faster development cycles compared to full process restarts.

packages/sanic-hmr/pyproject.toml

@@ -0,0 +1,31 @@
+[project]
+name = "sanic-hmr"
+description = "Hot Module Reloading for Sanic"
+version = "0.0.1"
+readme = "README.md"
+requires-python = ">=3.12"
+keywords = ["sanic", "hot-reload", "hmr", "reload", "server"]
+authors = [{ name = "Muspi Merol", email = "me@promplate.dev" }]
+license = "MIT"
+classifiers = [
+    "Development Status :: 4 - Beta",
+    "Intended Audience :: Developers",
+    "Operating System :: OS Independent",
+]
+dependencies = [
+    "dowhen~=0.1",
+    "hmr>=0.5.0,<0.7",
+    "typer-slim>=0.15.4,<1",
+    "sanic>=22.0.0",
+    "uvicorn>=0.24.0",
+]
+
+[project.scripts]
+sanic-hmr = "sanic_hmr:app"
+
+[project.urls]
+Homepage = "https://github.com/promplate/hmr"
+
+[build-system]
+requires = ["pdm-backend"]
+build-backend = "pdm.backend"

packages/sanic-hmr/sanic_hmr.py

@@ -0,0 +1,193 @@
+import sys
+from functools import cached_property
+from pathlib import Path
+from typing import TYPE_CHECKING, Annotated, override
+
+from typer import Argument, Option, Typer, secho
+
+app = Typer(help="Hot Module Replacement for Sanic", add_completion=False, pretty_exceptions_show_locals=False)
+
+
+@app.command(no_args_is_help=True)
+def main(
+    slug: Annotated[str, Argument()] = "main:app",
+    reload_include: list[str] = [str(Path.cwd())],  # noqa: B006, B008
+    reload_exclude: list[str] = [".venv"],  # noqa: B006
+    host: str = "localhost",
+    port: int = 8000,
+    workers: int = 1,  # noqa: ARG001  # Keep for CLI compatibility but force to 1
+    debug: Annotated[bool, Option("--debug", help="Enable debug mode")] = False,  # noqa: FBT002
+    access_log: Annotated[bool, Option("--access-log/--no-access-log", help="Enable/disable access log")] = True,  # noqa: FBT002
+    clear: Annotated[bool, Option("--clear", help="Clear the terminal before restarting the server")] = False,  # noqa: FBT002
+):
+    if ":" not in slug:
+        secho("Invalid slug: ", fg="red", nl=False)
+        secho(slug, fg="yellow")
+        exit(1)
+    module, attr = slug.split(":")
+
+    fragment = module.replace(".", "/")
+
+    file: Path | None
+    is_package = False
+    for path in ("", *sys.path):
+        if (file := Path(path, f"{fragment}.py")).is_file():
+            is_package = False
+            break
+        if (file := Path(path, fragment, "__init__.py")).is_file():
+            is_package = True
+            break
+    else:
+        file = None
+
+    if file is None:
+        secho("Module", fg="red", nl=False)
+        secho(f" {module} ", fg="yellow", nl=False)
+        secho("not found.", fg="red")
+        exit(1)
+
+    if module in sys.modules:
+        return secho(
+            f"It seems you've already imported `{module}` as a normal module. You should call `reactivity.hmr.core.patch_meta_path()` before it.",
+            fg="red",
+        )
+
+    from atexit import register
+    from importlib.machinery import ModuleSpec
+    from logging import getLogger
+    from threading import Event, Thread
+
+    from reactivity.hmr.core import ReactiveModule, ReactiveModuleLoader, SyncReloader, __version__, is_relative_to_any
+    from reactivity.hmr.utils import load
+    from watchfiles import Change
+
+    if TYPE_CHECKING:
+        from sanic import Sanic
+
+    cwd = str(Path.cwd())
+    if cwd not in sys.path:
+        sys.path.insert(0, cwd)
+
+    @register
+    def _():
+        stop_server()
+
+    def stop_server():
+        pass
+
+    def start_server(app: "Sanic"):
+        nonlocal stop_server
+
+        import uvicorn
+
+        finish = Event()
+        server = None
+
+        def run_server():
+            nonlocal server
+            watched_paths = [Path(p).resolve() for p in (file, *reload_include)]
+            ignored_paths = [Path(p).resolve() for p in reloader.excludes]
+            if all(is_relative_to_any(path, ignored_paths) or not is_relative_to_any(path, watched_paths) for path in ReactiveModule.instances):
+                logger.error("No files to watch for changes. The server will never reload.")
+
+            # Use Sanic as ASGI app with uvicorn server
+            # This avoids the threading/signal issues
+            config = uvicorn.Config(
+                app=app,  # Sanic app itself is the ASGI callable
+                host=host,
+                port=port,
+                log_level="info" if not debug else "debug",
+                access_log=access_log,
+            )
+            server = uvicorn.Server(config)
+            server.run()
+            finish.set()
+
+        Thread(target=run_server, daemon=True).start()
+
+        def stop_server():
+            if server:
+                server.should_exit = True
+            finish.wait()
+
+    class Reloader(SyncReloader):
+        def __init__(self):
+            super().__init__(str(file), reload_include, reload_exclude)
+            self.error_filter.exclude_filenames.add(__file__)  # exclude error stacks within this file
+
+        @cached_property
+        @override
+        def entry_module(self):
+            if "." in module:
+                __import__(module.rsplit(".", 1)[0])  # ensure parent modules are imported
+
+            if __version__ >= "0.6.4":
+                from reactivity.hmr.core import _loader as loader
+            else:
+                loader = ReactiveModuleLoader(file)  # type: ignore
+
+            spec = ModuleSpec(module, loader, origin=str(file), is_package=is_package)
+            sys.modules[module] = mod = loader.create_module(spec)
+            loader.exec_module(mod)
+            return mod
+
+        @override
+        def run_entry_file(self):
+            stop_server()
+            with self.error_filter:
+                load(self.entry_module)
+                app_instance = getattr(self.entry_module, attr)
+                # Ensure it's a Sanic app
+                if not hasattr(app_instance, "run"):
+                    secho(f"Attribute '{attr}' is not a Sanic app instance", fg="red")
+                    exit(1)
+                start_server(app_instance)
+
+        @override
+        def on_events(self, events):
+            if events:
+                paths: list[Path] = []
+                for type, file in events:
+                    path = Path(file).resolve()
+                    if type != Change.deleted and path in ReactiveModule.instances:
+                        paths.append(path)
+                if not paths:
+                    return
+
+                if clear:
+                    print("\033c", end="")
+                logger.warning("Watchfiles detected changes in %s. Reloading...", ", ".join(map(_display_path, paths)))
+            return super().on_events(events)
+
+        @override
+        def start_watching(self):
+            from dowhen import when
+
+            def log_server_restart():
+                logger.warning("Application '%s' has changed. Restarting server...", slug)
+
+            def log_module_reload(self: ReactiveModule):
+                ns = self.__dict__
+                logger.info("Reloading module '%s' from %s", ns["__name__"], _display_path(ns["__file__"]))
+
+            with (
+                when(ReactiveModule._ReactiveModule__load.method, "<start>").do(log_module_reload),  # type: ignore  # noqa: SLF001
+                when(self.run_entry_file, "<start>").do(log_server_restart),
+            ):
+                return super().start_watching()
+
+    logger = getLogger("sanic.error")
+    (reloader := Reloader()).keep_watching_until_interrupt()
+    stop_server()
+
+
+def _display_path(path: str | Path):
+    p = Path(path).resolve()
+    try:
+        return f"'{p.relative_to(Path.cwd())}'"
+    except ValueError:
+        return f"'{p}'"
+
+
+if __name__ == "__main__":
+    app()

pyproject.toml

@@ -7,6 +7,7 @@ dependencies = [
     "hmr~=0.6.0",
     "hmr-daemon",
     "ruff~=0.12.0",
+    "sanic-hmr",
     "uvicorn-hmr",
 ]
 
@@ -17,6 +18,7 @@ members = ["examples/*", "packages/*"]
 uvicorn-hmr = { workspace = true }
 fastapi-reloader = { workspace = true }
 hmr-daemon = { workspace = true }
+sanic-hmr = { workspace = true }
 
 [tool.pyright]
 typeCheckingMode = "standard"