from __future__ import annotations import asyncio from pathlib import Path from quart import Quart from .config import load_config from .database import init_db def create_app( config_path: Path | str | None = None, testing: bool = False, ) -> Quart: app = Quart(__name__, template_folder="templates") # Load config cfg = load_config(config_path) app.config.update( SECRET_KEY=cfg["secret_key"], DATABASE_URL=cfg["database"]["url"], SESSION_LIFETIME_HOURS=cfg.get("session", {}).get("lifetime_hours", 8), DATA_RETENTION_DAYS=cfg.get("data", {}).get("retention_days", 90), PLUGIN_DIR=cfg.get("plugin_dir", "plugins"), PLUGINS=cfg.get("plugins", {}), SMTP=cfg.get("smtp", {}), WEBHOOK=cfg.get("webhook", {}), ANSIBLE=cfg.get("ansible", {}), TESTING=testing, _RAW_CFG=cfg, ) # Database if not testing: init_db(app) else: from unittest.mock import MagicMock app.db_sessionmaker = MagicMock() # Run migrations (core + plugin) synchronously before event loop starts if not testing: from .core.migration_runner import run_migrations, get_plugin_migration_dirs _plugin_dir = Path(app.config["PLUGIN_DIR"]) _plugin_migration_dirs = get_plugin_migration_dirs( _plugin_dir, app.config["PLUGINS"] ) run_migrations(app.config["DATABASE_URL"], _plugin_migration_dirs) # Alert pipeline from .core.alerts import init_alerts init_alerts(app) # Register core blueprints from .auth.routes import auth_bp from .dashboard.routes import dashboard_bp from .hosts.routes import hosts_bp from .alerts.routes import alerts_bp from .ansible.routes import ansible_bp app.register_blueprint(auth_bp) app.register_blueprint(dashboard_bp) app.register_blueprint(hosts_bp) app.register_blueprint(alerts_bp) app.register_blueprint(ansible_bp) # Build task registry app._task_registry = [] if not testing: _register_core_tasks(app, cfg) # Load plugins if not testing: from .core.plugin_manager import load_plugins load_plugins(app) # Health check @app.get("/health") async def health(): return {"status": "ok"} # Start scheduler on first request @app.before_serving async def startup(): if not testing: await _mark_interrupted_runs(app) asyncio.create_task(_run_scheduler(app)) return app def _register_core_tasks(app: Quart, cfg: dict) -> None: from .core.scheduler import ScheduledTask poll_interval = cfg.get("monitors", {}).get("poll_interval_seconds", 60) cleanup_interval = 3600 # hourly async def run_ping_monitors(): from sqlalchemy import select from .models.hosts import Host from .monitors.ping import ping_check async with app.db_sessionmaker() as session: async with session.begin(): result = await session.execute( select(Host).where(Host.ping_enabled.is_(True)) ) for host in result.scalars().all(): try: await ping_check(host, session) except Exception: app.logger.exception(f"Ping check failed for host {host.name!r}") async def run_dns_monitors(): from sqlalchemy import select from .models.hosts import Host from .monitors.dns import dns_check async with app.db_sessionmaker() as session: async with session.begin(): result = await session.execute( select(Host).where(Host.dns_enabled.is_(True)) ) for host in result.scalars().all(): try: await dns_check(host, session) except Exception: app.logger.exception(f"DNS check failed for host {host.name!r}") async def run_cleanup(): from .core.cleanup import run_cleanup as _cleanup await _cleanup(app) app._task_registry.extend([ ScheduledTask( name="ping_monitor", coro_factory=run_ping_monitors, interval_seconds=poll_interval, run_on_startup=True, ), ScheduledTask( name="dns_monitor", coro_factory=run_dns_monitors, interval_seconds=poll_interval, run_on_startup=True, ), ScheduledTask( name="data_cleanup", coro_factory=run_cleanup, interval_seconds=cleanup_interval, run_on_startup=False, ), ]) # Register git-pull tasks for each git source ansible_cfg = cfg.get("ansible", {}) from .ansible.sources import get_sources for source in get_sources(ansible_cfg): if source["type"] != "git": continue _source = source # capture for closure async def run_git_pull(src=_source): from .ansible.sources import git_pull await git_pull(src) app._task_registry.append( ScheduledTask( name=f"ansible_git_pull_{_source['name']}", coro_factory=run_git_pull, interval_seconds=_source["pull_interval_seconds"], run_on_startup=True, ) ) async def _mark_interrupted_runs(app: Quart) -> None: """Mark any runs still in 'running' state as 'interrupted' (process restart).""" from sqlalchemy import update from .models.ansible import AnsibleRun, AnsibleRunStatus from datetime import datetime, timezone async with app.db_sessionmaker() as session: async with session.begin(): await session.execute( update(AnsibleRun) .where(AnsibleRun.status == AnsibleRunStatus.running) .values( status=AnsibleRunStatus.interrupted, finished_at=datetime.now(timezone.utc), ) ) async def _run_scheduler(app: Quart) -> None: from .core.scheduler import start_scheduler await start_scheduler(app._task_registry)