jtricerolph
4137813fc7
AI insights now fetches the Claude API key from the central settings service instead of storing its own encrypted copy, so wages and other apps can share the same key. Also wires up the update-available banner using the existing version-check hook pattern. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
"""
|
|
Client for the stack's central Settings service.
|
|
|
|
NewBook credentials are managed once in the Settings app (LXC 116) and
|
|
fetched live by every app — the same pattern as cashup / room-planner /
|
|
maintenance (see their lib/newbook.js). Falls back to None if the service
|
|
is unreachable so callers can fall back to app-local config.
|
|
"""
|
|
import logging
|
|
import os
|
|
import time
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SETTINGS_URL = os.getenv("SETTINGS_URL", "")
|
|
SETTINGS_SECRET = os.getenv("SETTINGS_SECRET", "")
|
|
|
|
_CACHE_TTL = 60 # seconds — credentials change rarely; avoid hammering the service
|
|
_cache: dict = {}
|
|
|
|
|
|
async def get_integration(name: str) -> Optional[dict]:
|
|
"""
|
|
Fetch integration config (e.g. 'newbook') from the central Settings
|
|
service. Returns the config dict, or None if unavailable/unconfigured.
|
|
"""
|
|
if not SETTINGS_URL or not SETTINGS_SECRET:
|
|
return None
|
|
|
|
cached = _cache.get(name)
|
|
if cached and (time.monotonic() - cached[0]) < _CACHE_TTL:
|
|
return cached[1]
|
|
|
|
url = f"{SETTINGS_URL}/settings/api/internal/integration/{name}"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5.0) as client:
|
|
resp = await client.get(
|
|
url, headers={"Authorization": f"Bearer {SETTINGS_SECRET}"}
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
_cache[name] = (time.monotonic(), data)
|
|
return data
|
|
except Exception as e:
|
|
logger.warning(f"Central settings fetch failed for '{name}': {e}")
|
|
return None
|
|
|
|
|
|
def _extract_newbook(s: Optional[dict]) -> Optional[dict]:
|
|
if not s:
|
|
return None
|
|
creds = {
|
|
"api_key": s.get("api_key") or "",
|
|
"username": s.get("username") or "",
|
|
"password": s.get("password") or "",
|
|
"region": s.get("region") or "eu",
|
|
}
|
|
# Only usable if the essential fields are present
|
|
if not (creds["api_key"] and creds["username"] and creds["password"]):
|
|
return None
|
|
return creds
|
|
|
|
|
|
async def get_newbook_credentials() -> Optional[dict]:
|
|
"""
|
|
Returns {'api_key', 'username', 'password', 'region'} from central
|
|
settings, or None if not available (caller should fall back).
|
|
"""
|
|
return _extract_newbook(await get_integration("newbook"))
|
|
|
|
|
|
def get_integration_sync(name: str) -> Optional[dict]:
|
|
"""Blocking variant of get_integration for sync job contexts."""
|
|
if not SETTINGS_URL or not SETTINGS_SECRET:
|
|
return None
|
|
|
|
cached = _cache.get(name)
|
|
if cached and (time.monotonic() - cached[0]) < _CACHE_TTL:
|
|
return cached[1]
|
|
|
|
url = f"{SETTINGS_URL}/settings/api/internal/integration/{name}"
|
|
try:
|
|
resp = httpx.get(
|
|
url, headers={"Authorization": f"Bearer {SETTINGS_SECRET}"}, timeout=5.0
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
_cache[name] = (time.monotonic(), data)
|
|
return data
|
|
except Exception as e:
|
|
logger.warning(f"Central settings fetch failed for '{name}': {e}")
|
|
return None
|
|
|
|
|
|
def get_newbook_credentials_sync() -> Optional[dict]:
|
|
"""Blocking variant of get_newbook_credentials for sync job contexts."""
|
|
return _extract_newbook(get_integration_sync("newbook"))
|
|
|
|
|
|
def _extract_resos(s: Optional[dict]) -> Optional[dict]:
|
|
if not s:
|
|
return None
|
|
key = s.get("api_key") or ""
|
|
if not key:
|
|
return None
|
|
return {"api_key": key}
|
|
|
|
|
|
async def get_resos_credentials() -> Optional[dict]:
|
|
"""Returns {'api_key'} from central settings, or None if not configured."""
|
|
return _extract_resos(await get_integration("resos"))
|
|
|
|
|
|
def get_resos_credentials_sync() -> Optional[dict]:
|
|
"""Blocking variant of get_resos_credentials for sync job contexts."""
|
|
return _extract_resos(get_integration_sync("resos"))
|
|
|
|
|
|
def _extract_anthropic(s: Optional[dict]) -> Optional[dict]:
|
|
if not s:
|
|
return None
|
|
key = s.get("api_key") or ""
|
|
if not key:
|
|
return None
|
|
return {"api_key": key}
|
|
|
|
|
|
async def get_anthropic_credentials() -> Optional[dict]:
|
|
"""Returns {'api_key'} from central settings, or None if not configured."""
|
|
return _extract_anthropic(await get_integration("anthropic"))
|