Spaces:
Sleeping
Sleeping
| import logging | |
| from pathlib import Path | |
| from typing import Optional, Dict, Any | |
| from config import DATABASE_CONFIG, BASE_DIR | |
| logger = logging.getLogger(__name__) | |
| from .db_adapter import DatabaseType | |
| def _get_database_type(): | |
| type_map = { | |
| "postgresql": DatabaseType.POSTGRESQL, | |
| "postgres": DatabaseType.POSTGRESQL, | |
| "mysql": DatabaseType.MYSQL, | |
| "mariadb": DatabaseType.MYSQL, | |
| "sqlite": DatabaseType.SQLITE, | |
| "mssql": DatabaseType.MSSQL, | |
| "sqlserver": DatabaseType.MSSQL, | |
| } | |
| db_type = DATABASE_CONFIG.get("type", "sqlite").lower() | |
| return type_map.get(db_type, DatabaseType.SQLITE) | |
| def init_database() -> Optional[Dict[str, Any]]: | |
| if not DATABASE_CONFIG.get("enabled", False): | |
| logger.info("Database integration is disabled") | |
| return None | |
| from .db_adapter import ConnectionConfig, get_database_adapter, reset_database_adapter | |
| try: | |
| reset_database_adapter() | |
| db_type = _get_database_type() | |
| if db_type == DatabaseType.SQLITE: | |
| db_path = DATABASE_CONFIG.get("path", "") | |
| if not db_path: | |
| return {"status": "error", "message": "SQLite path not set"} | |
| db_path_obj = Path(db_path) | |
| if not db_path_obj.is_absolute(): | |
| db_path_obj = BASE_DIR / db_path | |
| database_value = str(db_path_obj) | |
| else: | |
| database_value = DATABASE_CONFIG.get("name", "") | |
| config = ConnectionConfig( | |
| db_type=db_type, | |
| host=DATABASE_CONFIG.get("host", "localhost"), | |
| port=DATABASE_CONFIG.get("port", 5432), | |
| database=database_value, | |
| username=DATABASE_CONFIG.get("user", ""), | |
| password=DATABASE_CONFIG.get("password", ""), | |
| pool_min_size=DATABASE_CONFIG.get("pool_min_size", 2), | |
| pool_max_size=DATABASE_CONFIG.get("pool_max_size", 10), | |
| pool_timeout=DATABASE_CONFIG.get("pool_timeout", 30), | |
| query_timeout=DATABASE_CONFIG.get("query_timeout", 10), | |
| ) | |
| adapter = get_database_adapter(config) | |
| if not adapter.connect(): | |
| return {"status": "error", "message": "Database connection failed"} | |
| tables = [] | |
| try: | |
| rows = adapter.execute_query( | |
| "SELECT name FROM sqlite_master WHERE type IN ('table','view') AND name NOT LIKE 'sqlite_%' ORDER BY name" | |
| ) | |
| tables = [r["name"] for r in rows] | |
| except Exception: | |
| pass | |
| logger.info("Database integration initialized successfully") | |
| return {"status": "connected", "database_type": DATABASE_CONFIG.get("type"), "tables": tables} | |
| except Exception as e: | |
| logger.error(f"Database initialization failed: {e}") | |
| return {"status": "error", "message": str(e)} | |
| def shutdown_database() -> None: | |
| from .db_adapter import reset_database_adapter | |
| reset_database_adapter() | |
| logger.info("Database connections closed") | |
| def get_database_status() -> Dict[str, Any]: | |
| if not DATABASE_CONFIG.get("enabled", False): | |
| return {"status": "disabled"} | |
| from .db_adapter import get_database_adapter | |
| adapter = get_database_adapter() | |
| if adapter is None: | |
| return {"status": "not_initialized"} | |
| health = adapter.health_check() | |
| return {"status": "ready" if health.get("healthy") else "unhealthy", "connection": health} | |