middleware/app.py

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),
        }