Health Checks

The health namespace provides comprehensive system health monitoring capabilities. It includes basic health checks, detailed system status, and visualization service monitoring.

Basic Operations

Basic Health Check

from functor_sdk import FunctorClient

client = FunctorClient()

# Basic health check
health = client.health.check()
print(f"System status: {health['status']}")
print(f"System name: {health['system']}")
print(f"Version: {health['version']}")
print(f"Uptime: {health['uptime']}")

Detailed Health Check

# Detailed health check with component status
health = client.health.check_detailed(detailed=True)
print(f"Overall status: {health['status']}")
print(f"Components:")

for component, status in health['components'].items():
    print(f"  {component}: {status['status']}")
    if 'message' in status:
        print(f"    Message: {status['message']}")

Visualization Service Health

# Check visualization service specifically
viz_health = client.health.check_visualizations()
print(f"Visualization service status: {viz_health['status']}")
print(f"Service name: {viz_health['service']}")
print(f"Available endpoints: {viz_health['endpoints']}")

Complete Parameter Reference

check() Parameters

health = client.health.check()
# No parameters required

check_detailed() Parameters

health = client.health.check_detailed(
    detailed=True  # Optional: Include detailed component information
)

check_visualizations() Parameters

viz_health = client.health.check_visualizations()
# No parameters required

Response Formats

Basic Health Response

health = client.health.check()

# Basic properties
print(f"Status: {health['status']}")  # 'healthy', 'degraded', 'unhealthy'
print(f"System: {health['system']}")
print(f"Version: {health['version']}")
print(f"Uptime: {health['uptime']}")
print(f"Timestamp: {health['timestamp']}")

# Optional properties
if 'message' in health:
    print(f"Message: {health['message']}")
if 'checks' in health:
    print(f"Checks: {health['checks']}")

Detailed Health Response

health = client.health.check_detailed(detailed=True)

# Overall status
print(f"Overall status: {health['status']}")
print(f"System: {health['system']}")
print(f"Version: {health['version']}")

# Component status
for component, status in health['components'].items():
    print(f"\n{component}:")
    print(f"  Status: {status['status']}")
    if 'message' in status:
        print(f"  Message: {status['message']}")
    if 'details' in status:
        print(f"  Details: {status['details']}")
    if 'response_time_ms' in status:
        print(f"  Response time: {status['response_time_ms']}ms")

Visualization Health Response

viz_health = client.health.check_visualizations()

# Basic properties
print(f"Status: {viz_health['status']}")
print(f"Service: {viz_health['service']}")
print(f"Version: {viz_health['version']}")

# Available endpoints
if 'endpoints' in viz_health:
    print(f"Available endpoints: {viz_health['endpoints']}")

# Service details
if 'details' in viz_health:
    print(f"Service details: {viz_health['details']}")

Advanced Usage Patterns

Comprehensive Health Monitoring

def comprehensive_health_check():
    client = FunctorClient()
    
    health_report = {
        "timestamp": "2024-01-01T00:00:00Z",
        "overall_status": "unknown",
        "checks": {}
    }
    
    try:
        # Basic health check
        basic_health = client.health.check()
        health_report["checks"]["basic"] = {
            "status": basic_health["status"],
            "system": basic_health["system"],
            "version": basic_health["version"],
            "uptime": basic_health["uptime"]
        }
        
        # Detailed health check
        detailed_health = client.health.check_detailed(detailed=True)
        health_report["checks"]["detailed"] = {
            "status": detailed_health["status"],
            "components": detailed_health["components"]
        }
        
        # Visualization service health
        viz_health = client.health.check_visualizations()
        health_report["checks"]["visualizations"] = {
            "status": viz_health["status"],
            "service": viz_health["service"],
            "endpoints": viz_health.get("endpoints", [])
        }
        
        # Determine overall status
        statuses = [
            basic_health["status"],
            detailed_health["status"],
            viz_health["status"]
        ]
        
        if all(status == "healthy" for status in statuses):
            health_report["overall_status"] = "healthy"
        elif any(status == "unhealthy" for status in statuses):
            health_report["overall_status"] = "unhealthy"
        else:
            health_report["overall_status"] = "degraded"
        
    except Exception as e:
        health_report["overall_status"] = "error"
        health_report["error"] = str(e)
    
    return health_report

# Usage
report = comprehensive_health_check()
print(f"Overall status: {report['overall_status']}")
print(f"Checks performed: {len(report['checks'])}")

Async Health Monitoring

import asyncio
from functor_sdk import FunctorClient

async def async_health_monitoring():
    async with FunctorClient() as client:
        # Run all health checks concurrently
        tasks = [
            client.health.check_async(),
            client.health.check_detailed_async(detailed=True),
            client.health.check_visualizations_async()
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        basic_health, detailed_health, viz_health = results
        
        # Process results
        health_summary = {
            "basic_status": "error",
            "detailed_status": "error",
            "viz_status": "error"
        }
        
        if not isinstance(basic_health, Exception):
            health_summary["basic_status"] = basic_health["status"]
        
        if not isinstance(detailed_health, Exception):
            health_summary["detailed_status"] = detailed_health["status"]
        
        if not isinstance(viz_health, Exception):
            health_summary["viz_status"] = viz_health["status"]
        
        return health_summary

# Usage
summary = asyncio.run(async_health_monitoring())
print(f"Health Summary: {summary}")

Health Monitoring with Alerts

def health_monitoring_with_alerts():
    client = FunctorClient()
    
    alerts = []
    
    try:
        # Check basic health
        basic_health = client.health.check()
        if basic_health["status"] != "healthy":
            alerts.append(f"Basic health check failed: {basic_health['status']}")
        
        # Check detailed health
        detailed_health = client.health.check_detailed(detailed=True)
        for component, status in detailed_health["components"].items():
            if status["status"] != "healthy":
                alerts.append(f"Component {component} is {status['status']}")
                if "message" in status:
                    alerts.append(f"  Message: {status['message']}")
        
        # Check visualization service
        viz_health = client.health.check_visualizations()
        if viz_health["status"] != "healthy":
            alerts.append(f"Visualization service is {viz_health['status']}")
        
        # Check uptime
        uptime_hours = basic_health["uptime"] / 3600
        if uptime_hours < 1:
            alerts.append(f"System uptime is very low: {uptime_hours:.1f} hours")
        
    except Exception as e:
        alerts.append(f"Health check failed: {e}")
    
    return alerts

# Usage
alerts = health_monitoring_with_alerts()
if alerts:
    print("Health Alerts:")
    for alert in alerts:
        print(f"  - {alert}")
else:
    print("All health checks passed!")

Error Handling

Common Error Scenarios

from functor_sdk import (
    FunctorClient,
    FunctorAPIError,
    FunctorConnectionError,
    FunctorServiceUnavailableError
)

client = FunctorClient()

try:
    health = client.health.check()
    
except FunctorConnectionError:
    print("Connection failed - check your network and base URL")
    
except FunctorServiceUnavailableError:
    print("Service is currently unavailable")
    print("The system may be under maintenance")
    
except FunctorAPIError as e:
    print(f"API error {e.status_code}: {e.message}")
    
except Exception as e:
    print(f"Unexpected error: {e}")

Robust Health Checking

def robust_health_check(max_retries=3):
    client = FunctorClient()
    
    for attempt in range(max_retries):
        try:
            health = client.health.check()
            return {
                "status": "success",
                "health": health,
                "attempts": attempt + 1
            }
            
        except FunctorConnectionError as e:
            if attempt < max_retries - 1:
                print(f"Connection error, retrying... (attempt {attempt + 1})")
                time.sleep(2 ** attempt)  # Exponential backoff
            else:
                return {
                    "status": "connection_failed",
                    "error": str(e),
                    "attempts": max_retries
                }
                
        except FunctorServiceUnavailableError as e:
            return {
                "status": "service_unavailable",
                "error": str(e),
                "attempts": attempt + 1
            }
            
        except Exception as e:
            return {
                "status": "error",
                "error": str(e),
                "attempts": attempt + 1
            }
    
    return {
        "status": "max_retries_exceeded",
        "attempts": max_retries
    }

# Usage
result = robust_health_check()
if result["status"] == "success":
    print(f"Health check successful: {result['health']['status']}")
else:
    print(f"Health check failed: {result['status']}")

Performance Optimization

Optimization Tips

Use basic checks for frequent monitoring: Less overhead than detailed checks
Cache health results: Health status doesn't change frequently
Use async for concurrent checks: When monitoring multiple systems
Implement exponential backoff: For retry logic

Health Check Caching

import time
from functor_sdk import FunctorClient

class HealthCache:
    def __init__(self, client):
        self.client = client
        self._cache = {}
        self._cache_time = {}
        self.cache_duration = 60  # 1 minute
    
    def get_health(self, check_type="basic", force_refresh=False):
        now = time.time()
        
        # Check cache
        if not force_refresh and check_type in self._cache:
            if now - self._cache_time[check_type] < self.cache_duration:
                print(f"Using cached {check_type} health check")
                return self._cache[check_type]
        
        # Fetch fresh data
        print(f"Fetching fresh {check_type} health check")
        try:
            if check_type == "basic":
                health = self.client.health.check()
            elif check_type == "detailed":
                health = self.client.health.check_detailed(detailed=True)
            elif check_type == "visualizations":
                health = self.client.health.check_visualizations()
            else:
                raise ValueError(f"Unknown check type: {check_type}")
            
            # Update cache
            self._cache[check_type] = health
            self._cache_time[check_type] = now
            
            return health
            
        except Exception as e:
            return {"error": str(e)}
    
    def clear_cache(self):
        self._cache.clear()
        self._cache_time.clear()

# Usage
client = FunctorClient()
health_cache = HealthCache(client)

# First call - fetches from API
health1 = health_cache.get_health("basic")

# Second call - uses cache
health2 = health_cache.get_health("basic")

# Force refresh
health3 = health_cache.get_health("basic", force_refresh=True)

Integration Examples

System Status Dashboard

def create_system_dashboard():
    client = FunctorClient()
    
    dashboard = {
        "timestamp": "2024-01-01T00:00:00Z",
        "overall_status": "unknown",
        "components": {},
        "metrics": {}
    }
    
    try:
        # Basic health
        basic_health = client.health.check()
        dashboard["components"]["basic"] = {
            "status": basic_health["status"],
            "system": basic_health["system"],
            "version": basic_health["version"],
            "uptime": basic_health["uptime"]
        }
        
        # Detailed health
        detailed_health = client.health.check_detailed(detailed=True)
        dashboard["components"]["detailed"] = detailed_health["components"]
        
        # Visualization service
        viz_health = client.health.check_visualizations()
        dashboard["components"]["visualizations"] = {
            "status": viz_health["status"],
            "service": viz_health["service"],
            "endpoints": viz_health.get("endpoints", [])
        }
        
        # Calculate overall status
        statuses = [
            basic_health["status"],
            detailed_health["status"],
            viz_health["status"]
        ]
        
        if all(status == "healthy" for status in statuses):
            dashboard["overall_status"] = "healthy"
        elif any(status == "unhealthy" for status in statuses):
            dashboard["overall_status"] = "unhealthy"
        else:
            dashboard["overall_status"] = "degraded"
        
        # Add metrics
        dashboard["metrics"] = {
            "total_components": len(dashboard["components"]),
            "healthy_components": sum(1 for comp in dashboard["components"].values() 
                                    if isinstance(comp, dict) and comp.get("status") == "healthy"),
            "uptime_hours": basic_health["uptime"] / 3600
        }
        
    except Exception as e:
        dashboard["overall_status"] = "error"
        dashboard["error"] = str(e)
    
    return dashboard

# Usage
dashboard = create_system_dashboard()
print(f"System Dashboard:")
print(f"Overall Status: {dashboard['overall_status']}")
print(f"Components: {dashboard['metrics'].get('total_components', 0)}")
print(f"Healthy: {dashboard['metrics'].get('healthy_components', 0)}")

Health Monitoring Service

import time
import json
from datetime import datetime

class HealthMonitoringService:
    def __init__(self, client, check_interval=300):  # 5 minutes
        self.client = client
        self.check_interval = check_interval
        self.history = []
        self.max_history = 100
    
    def run_check(self):
        check_time = datetime.now().isoformat()
        
        try:
            # Perform health checks
            basic_health = self.client.health.check()
            detailed_health = self.client.health.check_detailed(detailed=True)
            viz_health = self.client.health.check_visualizations()
            
            # Record results
            health_record = {
                "timestamp": check_time,
                "basic_status": basic_health["status"],
                "detailed_status": detailed_health["status"],
                "viz_status": viz_health["status"],
                "uptime": basic_health["uptime"],
                "components": detailed_health["components"]
            }
            
            self.history.append(health_record)
            
            # Keep only recent history
            if len(self.history) > self.max_history:
                self.history = self.history[-self.max_history:]
            
            return health_record
            
        except Exception as e:
            error_record = {
                "timestamp": check_time,
                "error": str(e),
                "basic_status": "error",
                "detailed_status": "error",
                "viz_status": "error"
            }
            
            self.history.append(error_record)
            return error_record
    
    def get_status_summary(self):
        if not self.history:
            return {"status": "no_data"}
        
        recent = self.history[-1]
        
        return {
            "current_status": recent.get("basic_status", "unknown"),
            "last_check": recent["timestamp"],
            "uptime_hours": recent.get("uptime", 0) / 3600,
            "total_checks": len(self.history),
            "recent_errors": sum(1 for h in self.history[-10:] if "error" in h)
        }
    
    def export_history(self, filename=None):
        if filename is None:
            filename = f"health_history_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        
        with open(filename, 'w') as f:
            json.dump(self.history, f, indent=2)
        
        return filename

# Usage
client = FunctorClient()
monitor = HealthMonitoringService(client)

# Run a check
health_record = monitor.run_check()
print(f"Health check completed: {health_record['basic_status']}")

# Get summary
summary = monitor.get_status_summary()
print(f"Status summary: {summary}")

# Export history
filename = monitor.export_history()
print(f"History exported to: {filename}")

Next Steps

Knowledge Graphs Namespace - Monitor your knowledge graphs
Sources Namespace - Check source health
Visualizations Namespace - Monitor visualization services
Error Handling - Comprehensive error management