updates again

2025-09-08 17:42:02 -04:00
parent 678f84b6ef
commit e618b71620
3 changed files with 379 additions and 3 deletions
@@ -411,6 +411,141 @@ class RadarrDbClient:
            stats["error"] = str(e)
            
        return stats
+    
+    def health_check(self) -> Dict[str, Any]:
+        """
+        Comprehensive health check for the Radarr database connection
+        
+        Returns:
+            Dictionary with health status, connection info, and basic functionality tests
+        """
+        health = {
+            "status": "healthy",
+            "database_type": self.db_type,
+            "connection": "ok",
+            "readable": False,
+            "writable": False,
+            "tables_exist": False,
+            "sample_data": False,
+            "issues": [],
+            "tested_at": datetime.now(timezone.utc).isoformat(timespec="seconds")
+        }
+        
+        try:
+            # Test 1: Basic connection
+            with self._get_connection() as conn:
+                cursor = conn.cursor()
+                
+                # Test 2: Check if we can read (basic query)
+                try:
+                    cursor.execute('SELECT 1')
+                    result = cursor.fetchone()
+                    if result and result[0] == 1:
+                        health["readable"] = True
+                        health["connection"] = "readable"
+                    else:
+                        health["issues"].append("Basic SELECT query failed")
+                except Exception as e:
+                    health["issues"].append(f"Read test failed: {e}")
+                    health["status"] = "degraded"
+                
+                # Test 3: Check required tables exist
+                required_tables = ["Movies", "MovieMetadata", "History", "MovieFiles"]
+                existing_tables = []
+                
+                try:
+                    if self.db_type == "postgresql":
+                        cursor.execute("""
+                            SELECT table_name 
+                            FROM information_schema.tables 
+                            WHERE table_schema = 'public' 
+                            AND table_name IN ('Movies', 'MovieMetadata', 'History', 'MovieFiles')
+                        """)
+                    else:  # SQLite
+                        cursor.execute("""
+                            SELECT name 
+                            FROM sqlite_master 
+                            WHERE type='table' 
+                            AND name IN ('Movies', 'MovieMetadata', 'History', 'MovieFiles')
+                        """)
+                    
+                    rows = cursor.fetchall()
+                    existing_tables = [row[0] for row in rows]
+                    
+                    if len(existing_tables) == len(required_tables):
+                        health["tables_exist"] = True
+                    else:
+                        missing = set(required_tables) - set(existing_tables)
+                        health["issues"].append(f"Missing tables: {list(missing)}")
+                        health["status"] = "degraded"
+                        
+                    health["existing_tables"] = existing_tables
+                    
+                except Exception as e:
+                    health["issues"].append(f"Table check failed: {e}")
+                    health["status"] = "degraded"
+                
+                # Test 4: Check for sample data
+                if health["tables_exist"]:
+                    try:
+                        cursor.execute('SELECT COUNT(*) FROM "Movies"')
+                        movie_count = cursor.fetchone()[0]
+                        
+                        cursor.execute('SELECT COUNT(*) FROM "History"')
+                        history_count = cursor.fetchone()[0]
+                        
+                        if movie_count > 0 and history_count > 0:
+                            health["sample_data"] = True
+                            health["movie_count"] = movie_count
+                            health["history_count"] = history_count
+                        else:
+                            health["issues"].append(f"Low data counts - Movies: {movie_count}, History: {history_count}")
+                            
+                    except Exception as e:
+                        health["issues"].append(f"Sample data check failed: {e}")
+                
+                # Test 5: Test a real query (movie with IMDb lookup)
+                if health["sample_data"]:
+                    try:
+                        cursor.execute("""
+                            SELECT COUNT(*) 
+                            FROM "Movies" m
+                            JOIN "MovieMetadata" mm ON m."MovieMetadataId" = mm."Id"
+                            WHERE mm."ImdbId" IS NOT NULL
+                        """)
+                        imdb_movies = cursor.fetchone()[0]
+                        health["movies_with_imdb"] = imdb_movies
+                        
+                        if imdb_movies > 0:
+                            health["functional"] = True
+                        else:
+                            health["issues"].append("No movies with IMDb IDs found")
+                            
+                    except Exception as e:
+                        health["issues"].append(f"Functional test failed: {e}")
+                        health["status"] = "degraded"
+        
+        except Exception as e:
+            health["status"] = "error"
+            health["connection"] = "failed"
+            health["issues"].append(f"Connection failed: {e}")
+            _log("ERROR", f"Database health check failed: {e}")
+        
+        # Overall status determination
+        if health["issues"]:
+            if health["status"] == "healthy":
+                health["status"] = "degraded"
+        
+        # Add connection details (safe info only)
+        health["connection_info"] = {
+            "type": self.db_type,
+            "host": self.db_host if self.db_type == "postgresql" else None,
+            "port": self.db_port if self.db_type == "postgresql" else None,
+            "database": self.db_name if self.db_type == "postgresql" else None,
+            "path": self.db_path if self.db_type == "sqlite" else None
+        }
+        
+        return health


 if __name__ == "__main__":