Add watchfiles based reload class

setup.cfg

@@ -51,7 +51,7 @@ check_untyped_defs = True
 profile = black
 combine_as_imports = True
 known_first_party = uvicorn,tests
-known_third_party = click,does_not_exist,gunicorn,h11,httptools,pytest,requests,setuptools,urllib3,uvloop,watchgod,websockets,wsproto,yaml
+known_third_party = click,does_not_exist,gunicorn,h11,httptools,pytest,requests,setuptools,urllib3,uvloop,watchgod,watchfiles,websockets,wsproto,yaml
 
 [tool:pytest]
 addopts = -rxXs

setup.py

@@ -57,6 +57,7 @@ def get_packages(package):
     "uvloop>=0.14.0,!=0.15.0,!=0.15.1; " + env_marker_cpython,
     "colorama>=0.4;" + env_marker_win,
     "watchgod>=0.6",
+    "watchfiles>=0.10",
     "python-dotenv>=0.13",
     "PyYAML>=5.1",
 ]

uvicorn/supervisors/__init__.py

@@ -7,8 +7,15 @@
     ChangeReload: typing.Type[BaseReload]  # pragma: no cover
 else:
     try:
-        from uvicorn.supervisors.watchgodreload import WatchGodReload as ChangeReload
+        from uvicorn.supervisors.watchfilesreload import WatchFilesReload
+
+        ChangeReload = WatchFilesReload
     except ImportError:  # pragma: no cover
-        from uvicorn.supervisors.statreload import StatReload as ChangeReload
+        try:
+            from uvicorn.supervisors.watchgodreload import WatchGodReload
+
+            ChangeReload = WatchGodReload
+        except ImportError:  # pragma: no cover
+            from uvicorn.supervisors.statreload import StatReload as ChangeReload
 
 __all__ = ["Multiprocess", "ChangeReload"]

uvicorn/supervisors/watchfilesreload.py

@@ -0,0 +1,53 @@
+import logging
+from pathlib import Path
+from socket import socket
+from typing import TYPE_CHECKING
+
+from watchfiles import PythonFilter, watch
+
+from uvicorn.supervisors.basereload import BaseReload
+
+if TYPE_CHECKING:
+    from typing import Callable, List, Optional, Sequence, Union
+
+    from uvicorn.config import Config
+
+logger = logging.getLogger("uvicorn.error")
+
+
+class CustomFilter(PythonFilter):
+    def __init__(
+        self,
+        *,
+        ignore_paths: Optional[Sequence[Union[str, Path]]] = None,
+        extra_extensions: Sequence[str] = ()
+    ) -> None:
+        super().__init__(ignore_paths=ignore_paths, extra_extensions=extra_extensions)
+
+
+class WatchFilesReload(BaseReload):
+    def __init__(
+        self,
+        config: Config,
+        target: Callable[[Optional[List[socket]]], None],
+        sockets: List[socket],
+    ) -> None:
+        super().__init__(config, target, sockets)
+        self.reloader_name = "watchfiles"
+        self.reload_dirs = []
+
+        for directory in config.reload_dirs:
+            if Path.cwd() not in directory.parents:
+                self.reload_dirs.append(directory)
+        if Path.cwd() not in self.reload_dirs:
+            self.reload_dirs.append(Path.cwd())
+
+        self.watch_filter = CustomFilter()
+
+    def should_restart(self) -> bool:
+        for changes in watch(*self.reload_dirs, watch_filter=self.watch_filter):
+            message = "WatchFilesReload detected file change in '%s'. Reloading..."
+            logger.warning(message, [c[1] for c in changes])
+            return True
+
+        return False