137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
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", {}),
|
|
TESTING=testing,
|
|
_RAW_CFG=cfg,
|
|
)
|
|
|
|
# Database
|
|
if not testing:
|
|
init_db(app)
|
|
else:
|
|
from unittest.mock import MagicMock
|
|
app.db_sessionmaker = MagicMock()
|
|
|
|
# Alert pipeline
|
|
from .core.alerts import init_alerts
|
|
init_alerts(app)
|
|
|
|
# Register 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
|
|
|
|
app.register_blueprint(auth_bp)
|
|
app.register_blueprint(dashboard_bp)
|
|
app.register_blueprint(hosts_bp)
|
|
app.register_blueprint(alerts_bp)
|
|
|
|
# Build task registry
|
|
app._task_registry = []
|
|
|
|
if not testing:
|
|
_register_core_tasks(app, cfg)
|
|
|
|
# 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:
|
|
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,
|
|
),
|
|
])
|
|
|
|
|
|
async def _run_scheduler(app: Quart) -> None:
|
|
from .core.scheduler import start_scheduler
|
|
await start_scheduler(app._task_registry)
|