Spaces:
Paused
Paused
| #!/usr/bin/env python3 | |
| """ | |
| Complete Diagnostics Framework Integration Module | |
| """ | |
| import logging | |
| from core.diagnostics.simple_diagnostics import diagnostic_framework | |
| logger = logging.getLogger(__name__) | |
| class DiagnosticsIntegration: | |
| """Integration layer for diagnostics with main application""" | |
| def __init__(self): | |
| self.framework = diagnostic_framework() | |
| self.integration_active = False | |
| async def setup_diagnostics(self, app): | |
| """Setup diagnostics in FastAPI application""" | |
| try: | |
| success = await self.framework.setup_diagnostics(app) | |
| if success: | |
| self.integration_active = True | |
| logger.info( | |
| "✅ Diagnostics framework integrated successfully with FastAPI" | |
| ) | |
| print("📊 Diagnostics API endpoints are now available at:") | |
| print( | |
| " 🔍 /diagnostics/comprehensive - Complete diagnostics analysis" | |
| ) | |
| print(" 📊 /diagnostics/health - System health status") | |
| print(" 📊 /diagnostics/alerts - Recent alerts") | |
| print(" 📊 /diagnostics/summary - Dashboard overview") | |
| print(" 📊 /diagnostics/trends - Diagnostic trends") | |
| print(" 📊 /diagnostics/export - Export reports") | |
| print( | |
| " 📊 /diagnostics/continuous-monitoring/start - Start monitoring" | |
| ) | |
| print(" 📊 /diagnostics/continuous-monitoring/stop - Stop monitoring") | |
| print("") | |
| print("🎯 READY TO MONITOR YOUR COMPLETE APPLICATION!") | |
| else: | |
| logger.error("❌ Failed to integrate diagnostics framework") | |
| return success | |
| except Exception as e: | |
| logger.error(f"❌ Diagnostics integration failed: {e}") | |
| return False | |
| async def run_full_diagnostics(self) -> dict: | |
| """Run complete diagnostics and return results""" | |
| try: | |
| return await self.framework.run_comprehensive_diagnostics() | |
| except Exception as e: | |
| logger.error(f"❌ Failed to run comprehensive diagnostics: {e}") | |
| return {"error": str(e), "timestamp": self.framework.start_time.isoformat()} | |
| def get_latest_report(self) -> dict: | |
| """Get most recent diagnostic results""" | |
| try: | |
| return self.framework.get_summary() | |
| except Exception as e: | |
| return {"error": str(e), "timestamp": self.framework.start_time.isoformat()} | |
| async def start_continuous_monitoring(self, interval: int = 300) -> bool: | |
| """Start continuous monitoring""" | |
| return await self.framework.start_continuous_monitoring(interval=interval) | |
| async def stop_continuous_monitoring(self) -> bool: | |
| """Stop continuous monitoring""" | |
| return await self.framework.stop_continuous_monitoring() | |
| async def export_report(self, filename: str = None) -> bool: | |
| """Export latest diagnostic report""" | |
| try: | |
| report = self.get_latest_report() | |
| return await self.framework.export_report(report, filename) | |
| except Exception as e: | |
| logger.error(f"❌ Failed to export report: {e}") | |
| return False | |
| def get_api_router(self): | |
| """Get the diagnostics API router for integration""" | |
| return diagnostic_framework.router | |
| # Example usage in main.py: | |
| # from app.routers.diagnostics import DiagnosticsIntegration | |
| # | |
| # app = create_app() | |
| # diagnostics = DiagnosticsIntegration() | |
| # await diagnostics.setup_diagnostics(app) | |