feat: expose API runtime identity in health
This commit is contained in:
@@ -10,6 +10,7 @@ logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH
|
||||
from fastapi import FastAPI, HTTPException
|
||||
import uvicorn
|
||||
import requests
|
||||
from urllib.parse import urlparse
|
||||
from typing import Dict, Any, Optional
|
||||
from modules.database.tools.neo4j_driver_tools import get_driver
|
||||
|
||||
@@ -22,15 +23,59 @@ from modules.queue_system import ServiceType
|
||||
app = FastAPI()
|
||||
setup_cors(app)
|
||||
|
||||
def _truthy_env(name: str) -> bool:
|
||||
return os.getenv(name, "").lower() in {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
def _runtime_environment() -> str:
|
||||
"""Return the API runtime role used for dev/prod backing-service selection."""
|
||||
start_mode = os.getenv("START_MODE", "prod").lower()
|
||||
if start_mode == "dev" or _truthy_env("BACKEND_DEV_MODE"):
|
||||
return "dev"
|
||||
return "prod"
|
||||
|
||||
|
||||
def _url_host(url: Optional[str]) -> Optional[str]:
|
||||
if not url:
|
||||
return None
|
||||
parsed = urlparse(url)
|
||||
return parsed.hostname
|
||||
|
||||
|
||||
def _runtime_identity() -> Dict[str, Any]:
|
||||
"""Non-secret runtime identity for agents and smoke tests.
|
||||
|
||||
This intentionally exposes only modes, labels, and URL hosts. It must not
|
||||
include API keys, passwords, bearer tokens, or full URLs that may embed
|
||||
credentials.
|
||||
"""
|
||||
return {
|
||||
"api_runtime_role": _runtime_environment(),
|
||||
"start_mode": os.getenv("START_MODE", "prod"),
|
||||
"app_environment": os.getenv("APP_ENV"),
|
||||
"environment": os.getenv("ENVIRONMENT"),
|
||||
"backend_dev_mode": _truthy_env("BACKEND_DEV_MODE"),
|
||||
"compose_project": os.getenv("COMPOSE_PROJECT_NAME") or os.getenv("CC_COMPOSE_PROJECT"),
|
||||
"compose_service": os.getenv("COMPOSE_SERVICE") or os.getenv("CC_COMPOSE_SERVICE"),
|
||||
"supabase_url_host": _url_host(os.getenv("SUPABASE_URL")),
|
||||
}
|
||||
|
||||
|
||||
# Health check endpoint
|
||||
@app.get("/health")
|
||||
async def health_check() -> Dict[str, Any]:
|
||||
"""Health check endpoint that verifies all service dependencies"""
|
||||
runtime_identity = _runtime_identity()
|
||||
health_status = {
|
||||
"status": "healthy",
|
||||
"runtime": runtime_identity,
|
||||
"services": {
|
||||
"neo4j": {"status": "healthy", "message": "Connected"},
|
||||
"supabase": {"status": "healthy", "message": "Connected"},
|
||||
"supabase": {
|
||||
"status": "healthy",
|
||||
"message": "Connected",
|
||||
"url_host": runtime_identity["supabase_url_host"],
|
||||
},
|
||||
"redis": {"status": "healthy", "message": "Connected"}
|
||||
}
|
||||
}
|
||||
@@ -77,8 +122,8 @@ async def health_check() -> Dict[str, Any]:
|
||||
# Check Redis using new Redis manager
|
||||
from modules.redis_manager import get_redis_manager
|
||||
|
||||
# Determine environment
|
||||
environment = 'dev' if os.getenv('BACKEND_DEV_MODE', 'true').lower() == 'true' else 'prod'
|
||||
# Determine environment from explicit startup/runtime identity.
|
||||
environment = runtime_identity["api_runtime_role"]
|
||||
redis_manager = get_redis_manager(environment)
|
||||
|
||||
# Get comprehensive health check
|
||||
|
||||
Reference in New Issue
Block a user