"""Health check payload for /api/health.""" from __future__ import annotations from pathlib import Path from typing import Any, Dict, Optional, Tuple from flask import Flask def _project_root(app: Flask) -> Path: root = Path(__file__).resolve().parent.parent.parent ini = root / "alembic.ini" if ini.is_file(): return root return Path(app.config["BASE_DIR"]) def _alembic_head_revision(app: Flask) -> Optional[str]: try: from alembic.config import Config as AlembicConfig from alembic.script import ScriptDirectory root = _project_root(app) ini = root / "alembic.ini" if not ini.is_file(): return None cfg = AlembicConfig(str(ini)) cfg.set_main_option("script_location", str(root / "migrations")) return ScriptDirectory.from_config(cfg).get_current_head() except Exception: return None def _db_migration_revision(app: Flask) -> Tuple[Optional[str], bool]: """(текущая ревизия в recipes.db, таблица alembic_version есть).""" try: from sqlalchemy import inspect, text from app import db insp = inspect(db.engine) if not insp.has_table("alembic_version"): return None, False rev = db.session.execute(text("SELECT version_num FROM alembic_version LIMIT 1")).scalar() return (str(rev) if rev is not None else None), True except Exception: try: from app import db db.session.rollback() except Exception: pass return None, False def migrations_ok_for_app(app: Flask) -> bool: if app.config.get("TESTING"): return True head = _alembic_head_revision(app) current, has_table = _db_migration_revision(app) if head is None: return has_table if not has_table: return False return current == head def build_health_payload(app: Flask) -> Dict[str, Any]: ver = str(app.config.get("SYNC_CLIENT_VERSION", "2.0.0") or "2.0.0") from app.services.startup_background import is_startup_ready_implicit from app.services.startup_state import startup_status_payload if not is_startup_ready_implicit(app): st = startup_status_payload() return { "ok": False, "starting": True, "version": ver, "db_ok": False, "migrations_ok": False, **st, } db_ok = True migrations_ok = True migration_revision: Optional[str] = None migration_head: Optional[str] = None if not app.config.get("TESTING"): try: from app import db from sqlalchemy import text db.session.execute(text("SELECT 1")) db.session.commit() except Exception: db_ok = False try: from app import db db.session.rollback() except Exception: pass migration_head = _alembic_head_revision(app) migration_revision, _ = _db_migration_revision(app) migrations_ok = migrations_ok_for_app(app) ok = db_ok and migrations_ok payload: Dict[str, Any] = { "ok": ok, "version": ver, "db_ok": db_ok, "migrations_ok": migrations_ok, } if migration_revision is not None: payload["migration_revision"] = migration_revision if migration_head is not None: payload["migration_head"] = migration_head return payload