Add a reloader to Quart

This is based on the old Hypercorn reloader, and is included as the
Quart reloader must run in the same process and event loop as the
app. This is required as the app is often adapted before `run` is
called, hence the new Hypercorn process method will break
expectations.

This is a slight shame, as the new Hypercorn reloader is much more
reliable, however it does keep compatibility and is acceptable given
the strong reminders that `run` is for development only.

Author: pgjones

src/quart/app.py

@@ -126,7 +126,14 @@
     TestClientProtocol,
     WhileServingCallable,
 )
-from .utils import file_path_to_path, is_coroutine_function, run_sync
+from .utils import (
+    file_path_to_path,
+    is_coroutine_function,
+    MustReloadError,
+    observe_changes,
+    restart,
+    run_sync,
+)
 from .wrappers import BaseRequestWebsocket, Request, Response, Websocket
 
 AppOrBlueprintKey = Optional[str]  # The App key is None, whereas blueprints are named
@@ -1386,7 +1393,6 @@ def _signal_handler(*_: Any) -> None:
             host,
             port,
             debug,
-            use_reloader,
             ca_certs,
             certfile,
             keyfile,
@@ -1402,8 +1408,15 @@ def _signal_handler(*_: Any) -> None:
         scheme = "https" if certfile is not None and keyfile is not None else "http"
         print(f" * Running on {scheme}://{host}:{port} (CTRL + C to quit)")  # noqa: T201
 
+        tasks = [loop.create_task(task)]
+        if use_reloader:
+            tasks.append(loop.create_task(observe_changes(asyncio.sleep, shutdown_event)))
+
+        reload_ = False
         try:
-            loop.run_until_complete(task)
+            loop.run_until_complete(asyncio.gather(*tasks))
+        except MustReloadError:
+            reload_ = True
         finally:
             try:
                 _cancel_all_tasks(loop)
@@ -1412,12 +1425,14 @@ def _signal_handler(*_: Any) -> None:
                 asyncio.set_event_loop(None)
                 loop.close()
 
+        if reload_:
+            restart()
+
     def run_task(
         self,
         host: str = "127.0.0.1",
         port: int = 5000,
         debug: Optional[bool] = None,
-        use_reloader: bool = True,
         ca_certs: Optional[str] = None,
         certfile: Optional[str] = None,
         keyfile: Optional[str] = None,
@@ -1433,7 +1448,6 @@ def run_task(
                 only, use 0.0.0.0 to have the server listen externally.
             port: Port number to listen on.
             debug: If set enable (or disable) debug mode and debug output.
-            use_reloader: Automatically reload on code changes.
             ca_certs: Path to the SSL CA certificate file.
             certfile: Path to the SSL certificate file.
             keyfile: Path to the SSL key file.
@@ -1449,7 +1463,6 @@ def run_task(
             self.debug = debug
         config.errorlog = config.accesslog
         config.keyfile = keyfile
-        config.use_reloader = use_reloader
 
         return serve(self, config, shutdown_trigger=shutdown_trigger)
 

src/quart/typing.py

@@ -340,3 +340,8 @@ async def __aenter__(self) -> TestAppProtocol:
 
     async def __aexit__(self, exc_type: type, exc_value: BaseException, tb: TracebackType) -> None:
         ...
+
+
+class Event(Protocol):
+    def is_set(self) -> bool:
+        ...

src/quart/utils.py

@@ -2,16 +2,19 @@
 
 import asyncio
 import inspect
+import os
+import platform
 import sys
 from contextvars import copy_context
 from functools import partial, wraps
-from os import PathLike
 from pathlib import Path
 from typing import (
     Any,
     AsyncGenerator,
+    Awaitable,
     Callable,
     Coroutine,
+    Dict,
     Generator,
     Iterable,
     List,
@@ -22,15 +25,19 @@
 
 from werkzeug.datastructures import Headers
 
-from .typing import FilePath
+from .typing import Event, FilePath
 
 if TYPE_CHECKING:
     from .wrappers.response import Response  # noqa: F401
 
 
+class MustReloadError(Exception):
+    pass
+
+
 def file_path_to_path(*paths: FilePath) -> Path:
     # Flask supports bytes paths
-    safe_paths: List[Union[str, PathLike]] = []
+    safe_paths: List[Union[str, os.PathLike]] = []
     for path in paths:
         if isinstance(path, bytes):
             safe_paths.append(path.decode())
@@ -122,3 +129,69 @@ def encode_headers(headers: Headers) -> List[Tuple[bytes, bytes]]:
 
 def decode_headers(headers: Iterable[Tuple[bytes, bytes]]) -> Headers:
     return Headers([(key.decode(), value.decode()) for key, value in headers])
+
+
+async def observe_changes(sleep: Callable[[float], Awaitable[Any]], shutdown_event: Event) -> None:
+    last_updates: Dict[Path, float] = {}
+    for module in list(sys.modules.values()):
+        filename = getattr(module, "__file__", None)
+        if filename is None:
+            continue
+        path = Path(filename)
+        try:
+            last_updates[Path(filename)] = path.stat().st_mtime
+        except (FileNotFoundError, NotADirectoryError):
+            pass
+
+    while not shutdown_event.is_set():
+        await sleep(1)
+
+        for index, (path, last_mtime) in enumerate(last_updates.items()):
+            if index % 10 == 0:
+                # Yield to the event loop
+                await sleep(0)
+
+            try:
+                mtime = path.stat().st_mtime
+            except FileNotFoundError:
+                # File deleted
+                raise MustReloadError()
+            else:
+                if mtime > last_mtime:
+                    raise MustReloadError()
+                else:
+                    last_updates[path] = mtime
+
+
+def restart() -> None:
+    # Restart  this process (only safe for dev/debug)
+    executable = sys.executable
+    script_path = Path(sys.argv[0]).resolve()
+    args = sys.argv[1:]
+    main_package = sys.modules["__main__"].__package__
+
+    if main_package is None:
+        # Executed by filename
+        if platform.system() == "Windows":
+            if not script_path.exists() and script_path.with_suffix(".exe").exists():
+                # quart run
+                executable = str(script_path.with_suffix(".exe"))
+            else:
+                # python run.py
+                args.append(str(script_path))
+        else:
+            if script_path.is_file() and os.access(script_path, os.X_OK):
+                # hypercorn run:app --reload
+                executable = str(script_path)
+            else:
+                # python run.py
+                args.append(str(script_path))
+    else:
+        # Executed as a module e.g. python -m run
+        module = script_path.stem
+        import_name = main_package
+        if module != "__main__":
+            import_name = f"{main_package}.{module}"
+        args[:0] = ["-m", import_name.lstrip(".")]
+
+    os.execv(executable, [executable] + args)