#!/usr/bin/env python3
"""
Debugger Database - 디버거 데이터베이스 관리
디버깅 세션, 브레이크포인트, 이벤트 로그 등의 데이터베이스 관리
"""
import sqlite3
import logging
from typing import Dict, List, Optional, Any
from datetime import datetime
from .models import DebugSession, BreakpointInfo, DebugEvent
logger = logging.getLogger(__name__)
class MCPDebuggerDatabase:
"""MCP Debugger 데이터베이스 관리자"""
def __init__(self, db_path: str = "mcp_debugger.db"):
"""
데이터베이스 초기화
Args:
db_path: 데이터베이스 파일 경로
"""
self.db_path = db_path
self.init_database()
def init_database(self):
"""데이터베이스 초기화 및 테이블 생성"""
try:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
# 디버깅 세션 테이블
cursor.execute("""
CREATE TABLE IF NOT EXISTS debug_sessions (
session_id TEXT PRIMARY KEY,
language TEXT NOT NULL,
name TEXT NOT NULL,
state TEXT NOT NULL,
executable_path TEXT,
script_path TEXT,
dap_port INTEGER,
timeout_seconds INTEGER DEFAULT 300,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_activity TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# 브레이크포인트 테이블
cursor.execute("""
CREATE TABLE IF NOT EXISTS breakpoints (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
file_path TEXT NOT NULL,
line INTEGER NOT NULL,
condition TEXT,
enabled BOOLEAN DEFAULT TRUE,
hit_count INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (session_id) REFERENCES debug_sessions (session_id)
)
""")
# 디버깅 이벤트 로그 테이블
cursor.execute("""
CREATE TABLE IF NOT EXISTS debug_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
event_type TEXT NOT NULL,
event_data TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (session_id) REFERENCES debug_sessions (session_id)
)
""")
# 성능 메트릭 테이블
cursor.execute("""
CREATE TABLE IF NOT EXISTS debug_metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
metric_name TEXT NOT NULL,
metric_value REAL NOT NULL,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (session_id) REFERENCES debug_sessions (session_id)
)
""")
# 인덱스 생성
cursor.execute("CREATE INDEX IF NOT EXISTS idx_session_language ON debug_sessions(language)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_breakpoint_session ON breakpoints(session_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_event_session ON debug_events(session_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_metric_session ON debug_metrics(session_id)")
conn.commit()
logger.info("🔧 MCP Debugger 데이터베이스 초기화 완료")
except Exception as e:
logger.error(f"❌ 데이터베이스 초기화 실패: {e}")
raise
def save_session(self, session: DebugSession):
"""세션 정보 저장"""
try:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO debug_sessions
(session_id, language, name, state, executable_path, created_at, last_activity)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
session.session_id,
session.language.value,
session.name,
session.state.value,
session.executable_path,
session.created_at.isoformat(),
session.last_activity.isoformat()
))
conn.commit()
except Exception as e:
logger.error(f"세션 저장 실패: {e}")
def get_session(self, session_id: str) -> Optional[DebugSession]:
"""세션 조회"""
try:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT session_id, language, name, state, executable_path, created_at, last_activity
FROM debug_sessions WHERE session_id = ?
""", (session_id,))
row = cursor.fetchone()
if row:
from .models import DebuggerLanguage, DebuggerState
return DebugSession(
session_id=row[0],
language=DebuggerLanguage(row[1]),
name=row[2],
state=DebuggerState(row[3]),
executable_path=row[4],
created_at=datetime.fromisoformat(row[5]),
last_activity=datetime.fromisoformat(row[6])
)
except Exception as e:
logger.error(f"세션 조회 실패: {e}")
return None
def save_breakpoint(self, breakpoint: BreakpointInfo, session_id: str):
"""브레이크포인트 저장"""
try:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO breakpoints
(id, session_id, file_path, line, condition, enabled, hit_count, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
breakpoint.id,
session_id,
breakpoint.file_path,
breakpoint.line,
breakpoint.condition,
breakpoint.enabled,
breakpoint.hit_count,
breakpoint.created_at.isoformat()
))
conn.commit()
except Exception as e:
logger.error(f"브레이크포인트 저장 실패: {e}")
def get_breakpoints(self, session_id: str) -> List[BreakpointInfo]:
"""세션의 브레이크포인트 목록 조회"""
breakpoints = []
try:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT id, file_path, line, condition, enabled, hit_count, created_at
FROM breakpoints WHERE session_id = ?
""", (session_id,))
for row in cursor.fetchall():
breakpoints.append(BreakpointInfo(
id=row[0],
file_path=row[1],
line=row[2],
condition=row[3],
enabled=bool(row[4]),
hit_count=row[5],
created_at=datetime.fromisoformat(row[6])
))
except Exception as e:
logger.error(f"브레이크포인트 조회 실패: {e}")
return breakpoints
def log_event(self, event: DebugEvent):
"""디버그 이벤트 로깅"""
try:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO debug_events (session_id, event_type, event_data, timestamp)
VALUES (?, ?, ?, ?)
""", (
event.session_id,
event.event_type,
str(event.data),
event.timestamp.isoformat()
))
conn.commit()
except Exception as e:
logger.error(f"이벤트 로깅 실패: {e}")
def save_metric(self, session_id: str, metric_name: str, metric_value: float):
"""성능 메트릭 저장"""
try:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO debug_metrics (session_id, metric_name, metric_value)
VALUES (?, ?, ?)
""", (session_id, metric_name, metric_value))
conn.commit()
except Exception as e:
logger.error(f"메트릭 저장 실패: {e}")
def cleanup_old_sessions(self, hours: int = 24):
"""오래된 세션 정리"""
try:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
# 오래된 세션들 찾기
cursor.execute("""
SELECT session_id FROM debug_sessions
WHERE last_activity < datetime('now', '-{} hours')
""".format(hours))
old_sessions = [row[0] for row in cursor.fetchall()]
# 관련 데이터 삭제
for session_id in old_sessions:
cursor.execute("DELETE FROM debug_events WHERE session_id = ?", (session_id,))
cursor.execute("DELETE FROM debug_metrics WHERE session_id = ?", (session_id,))
cursor.execute("DELETE FROM breakpoints WHERE session_id = ?", (session_id,))
cursor.execute("DELETE FROM debug_sessions WHERE session_id = ?", (session_id,))
conn.commit()
if old_sessions:
logger.info(f"정리된 오래된 세션: {len(old_sessions)}개")
except Exception as e:
logger.error(f"세션 정리 실패: {e}")
def get_session_metrics(self, session_id: str) -> Dict[str, List[float]]:
"""세션의 성능 메트릭 조회"""
metrics = {}
try:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT metric_name, metric_value FROM debug_metrics
WHERE session_id = ? ORDER BY timestamp
""", (session_id,))
for metric_name, metric_value in cursor.fetchall():
if metric_name not in metrics:
metrics[metric_name] = []
metrics[metric_name].append(metric_value)
except Exception as e:
logger.error(f"메트릭 조회 실패: {e}")
return metrics
