Spaces:
Sleeping
Sleeping
File size: 3,431 Bytes
79d4fd5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | 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}
|