File size: 2,265 Bytes
db521fa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | import logging
import time
from typing import Any, Optional
logger = logging.getLogger(__name__)
_MAX_RETRIES = 3
class App:
def __init__(self, debug: bool = False, dry_run: bool = False) -> None:
self.debug = debug
self.dry_run = dry_run
self._started = False
self._components: list[Any] = []
self._hooks: dict[str, list] = {"startup": [], "shutdown": []}
def on_startup(self, fn) -> None:
self._hooks["startup"].append(fn)
def on_shutdown(self, fn) -> None:
self._hooks["shutdown"].append(fn)
def _run_hooks(self, event: str) -> None:
for fn in self._hooks.get(event, []):
try:
fn()
except Exception as exc:
logger.error("Hook %s raised: %s", fn.__name__, exc)
def _load_components(self) -> None:
logger.debug("Loading components")
self._components = []
def _warmup(self) -> bool:
for attempt in range(1, _MAX_RETRIES + 1):
try:
self._load_components()
return True
except Exception as exc:
logger.warning("Warmup attempt %d/%d: %s", attempt, _MAX_RETRIES, exc)
time.sleep(0.5 * attempt)
return False
def run(self) -> None:
logger.info("Starting (debug=%s dry_run=%s)", self.debug, self.dry_run)
if not self._warmup():
logger.error("Warmup failed, aborting")
return
self._started = True
self._run_hooks("startup")
try:
self._main_loop()
finally:
self.shutdown()
def _main_loop(self) -> None:
if self.dry_run:
logger.info("[dry-run] skipping main loop")
return
logger.info("Running main loop")
def shutdown(self) -> None:
if not self._started:
return
self._run_hooks("shutdown")
self._components.clear()
self._started = False
logger.info("Shutdown complete")
def status(self) -> dict:
return {
"started": self._started,
"debug": self.debug,
"dry_run": self.dry_run,
"components": len(self._components),
}
|