r_uvicorn_worker.py 947 B

123456789101112131415161718192021222324252627282930313233343536
  1. import os
  2. import signal
  3. import threading
  4. import time
  5. from typing import Any, List, Dict
  6. from uvicorn.workers import UvicornWorker
  7. class ReloaderThread(threading.Thread):
  8. def __init__(self, worker: UvicornWorker, sleep_interval: float = 1.0):
  9. super().__init__()
  10. self.setDaemon(True)
  11. self._worker = worker
  12. self._interval = sleep_interval
  13. def run(self) -> None:
  14. while True:
  15. if not self._worker.alive:
  16. os.kill(os.getpid(), signal.SIGINT)
  17. time.sleep(self._interval)
  18. class RestartableUvicornWorker(UvicornWorker):
  19. CONFIG_KWARGS = {"loop": "uvloop", "http": "httptools"}
  20. def __init__(self, *args: List[Any], **kwargs: Dict[str, Any]):
  21. super().__init__(*args, **kwargs)
  22. self._reloader_thread = ReloaderThread(self)
  23. def run(self) -> None:
  24. if self.cfg.reload:
  25. self._reloader_thread.start()
  26. super().run()