Spaces:
Sleeping
Sleeping
File size: 5,256 Bytes
f871fed | 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 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | import asyncio
import os
import time
import tomllib
from pathlib import Path
from typing import Optional
from fastapi import APIRouter, Request
from loguru import logger
from open_notebook.database.repository import repo_query
from open_notebook.utils.version_utils import (
compare_versions,
get_version_from_github,
)
router = APIRouter()
# In-memory cache for version check results
_version_cache: dict = {
"latest_version": None,
"has_update": False,
"timestamp": 0,
"check_failed": False,
}
# Cache TTL in seconds (24 hours)
VERSION_CACHE_TTL = 24 * 60 * 60
def get_version() -> str:
"""Read version from pyproject.toml"""
try:
pyproject_path = Path(__file__).parent.parent.parent / "pyproject.toml"
with open(pyproject_path, "rb") as f:
pyproject = tomllib.load(f)
return pyproject.get("project", {}).get("version", "unknown")
except Exception as e:
logger.warning(f"Could not read version from pyproject.toml: {e}")
return "unknown"
def get_latest_version_cached(current_version: str) -> tuple[Optional[str], bool]:
"""
Check for the latest version from GitHub with caching.
Returns:
tuple: (latest_version, has_update)
- latest_version: str or None if check failed
- has_update: bool indicating if update is available
"""
global _version_cache
# Check if cache is still valid (within TTL)
cache_age = time.time() - _version_cache["timestamp"]
if _version_cache["timestamp"] > 0 and cache_age < VERSION_CACHE_TTL:
logger.debug(f"Using cached version check result (age: {cache_age:.0f}s)")
return _version_cache["latest_version"], _version_cache["has_update"]
# Cache expired or not yet set
if _version_cache["timestamp"] > 0:
logger.info(f"Version cache expired (age: {cache_age:.0f}s), refreshing...")
# Perform version check with strict error handling
try:
logger.info("Checking for latest version from GitHub...")
# Fetch latest version from GitHub with 10-second timeout
latest_version = get_version_from_github(
"https://github.com/lfnovo/open-notebook",
"main"
)
logger.info(f"Latest version from GitHub: {latest_version}, Current version: {current_version}")
# Compare versions
has_update = compare_versions(current_version, latest_version) < 0
# Cache the result
_version_cache["latest_version"] = latest_version
_version_cache["has_update"] = has_update
_version_cache["timestamp"] = time.time()
_version_cache["check_failed"] = False
logger.info(f"Version check complete. Update available: {has_update}")
return latest_version, has_update
except Exception as e:
logger.warning(f"Version check failed: {e}")
# Cache the failure to avoid repeated attempts
_version_cache["latest_version"] = None
_version_cache["has_update"] = False
_version_cache["timestamp"] = time.time()
_version_cache["check_failed"] = True
return None, False
async def check_database_health() -> dict:
"""
Check if database is reachable using a lightweight query.
Returns:
dict with 'status' ("online" | "offline") and optional 'error'
"""
try:
# 2-second timeout for database health check
result = await asyncio.wait_for(
repo_query("RETURN 1"),
timeout=2.0
)
if result:
return {"status": "online"}
return {"status": "offline", "error": "Empty result"}
except asyncio.TimeoutError:
logger.warning("Database health check timed out after 2 seconds")
return {"status": "offline", "error": "Health check timeout"}
except Exception as e:
logger.warning(f"Database health check failed: {e}")
return {"status": "offline", "error": str(e)}
@router.get("/config")
async def get_config(request: Request):
"""
Get frontend configuration.
Returns version information and health status.
Note: The frontend determines the API URL via its own runtime-config endpoint,
so this endpoint no longer returns apiUrl.
Also checks for version updates from GitHub (with caching and error handling).
"""
# Get current version
current_version = get_version()
# Check for updates (with caching and error handling)
# This MUST NOT break the endpoint - wrapped in try-except as extra safety
latest_version = None
has_update = False
try:
latest_version, has_update = get_latest_version_cached(current_version)
except Exception as e:
# Extra safety: ensure version check never breaks the config endpoint
logger.error(f"Unexpected error during version check: {e}")
# Check database health
db_health = await check_database_health()
db_status = db_health["status"]
if db_status == "offline":
logger.warning(f"Database offline: {db_health.get('error', 'Unknown error')}")
return {
"version": current_version,
"latestVersion": latest_version,
"hasUpdate": has_update,
"dbStatus": db_status,
}
|