8adeece5c3
- Wrap PRAGMA statement with text() for SQLAlchemy 2.0 - Fixes ObjectNotExecutableError on database initialization
134 lines
3.9 KiB
Python
134 lines
3.9 KiB
Python
"""Database configuration and models for Claude Code Monitor."""
|
|
|
|
from sqlalchemy import (
|
|
create_engine,
|
|
Column,
|
|
Integer,
|
|
String,
|
|
Float,
|
|
Boolean,
|
|
DateTime,
|
|
Index,
|
|
CheckConstraint,
|
|
text,
|
|
)
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.sql import func
|
|
import os
|
|
|
|
# Database URL
|
|
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./data/claude_monitor.db")
|
|
|
|
# Create engine with WAL mode for better concurrency
|
|
engine = create_engine(
|
|
DATABASE_URL,
|
|
connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {},
|
|
echo=False,
|
|
)
|
|
|
|
# Session factory
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
# Base class for models
|
|
Base = declarative_base()
|
|
|
|
|
|
class Event(Base):
|
|
"""Event model for storing Claude Code events."""
|
|
|
|
__tablename__ = "events"
|
|
|
|
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
|
|
session_id = Column(String, nullable=False, index=True)
|
|
event_type = Column(String, nullable=False, index=True) # PreToolUse, PostToolUse, etc.
|
|
tool_name = Column(String, nullable=True, index=True) # Bash, Read, Write, etc.
|
|
tool_input = Column(String, nullable=True) # JSON string
|
|
tool_output = Column(String, nullable=True) # Tool response
|
|
success = Column(Boolean, nullable=True) # Success/failure (NULL for PreToolUse)
|
|
description = Column(String, nullable=True) # Human-readable description
|
|
timestamp = Column(Float, nullable=False, index=True) # Unix timestamp
|
|
created_at = Column(DateTime, server_default=func.now())
|
|
|
|
__table_args__ = (
|
|
Index("idx_event_type_timestamp", "event_type", "timestamp"),
|
|
Index("idx_session_timestamp", "session_id", "timestamp"),
|
|
)
|
|
|
|
|
|
class Session(Base):
|
|
"""Session model for tracking Claude Code sessions."""
|
|
|
|
__tablename__ = "sessions"
|
|
|
|
session_id = Column(String, primary_key=True)
|
|
started_at = Column(Float, nullable=False, index=True)
|
|
ended_at = Column(Float, nullable=True)
|
|
event_count = Column(Integer, default=0)
|
|
created_at = Column(DateTime, server_default=func.now())
|
|
|
|
|
|
class Statistics(Base):
|
|
"""Statistics model for cached aggregations (single-row table)."""
|
|
|
|
__tablename__ = "statistics"
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
total_events = Column(Integer, default=0)
|
|
total_tools_used = Column(Integer, default=0)
|
|
total_agents = Column(Integer, default=0)
|
|
total_sessions = Column(Integer, default=0)
|
|
last_updated = Column(Float, nullable=False)
|
|
|
|
__table_args__ = (CheckConstraint("id = 1", name="single_row"),)
|
|
|
|
|
|
class ToolUsage(Base):
|
|
"""Tool usage model for tracking unique tools."""
|
|
|
|
__tablename__ = "tool_usage"
|
|
|
|
tool_name = Column(String, primary_key=True)
|
|
usage_count = Column(Integer, default=1)
|
|
first_used = Column(Float, nullable=False)
|
|
last_used = Column(Float, nullable=False, index=True)
|
|
|
|
|
|
def init_db():
|
|
"""Initialize database: create tables and enable WAL mode."""
|
|
# Create all tables
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
# Enable WAL mode for SQLite
|
|
if "sqlite" in DATABASE_URL:
|
|
with engine.connect() as conn:
|
|
conn.execute(text("PRAGMA journal_mode=WAL"))
|
|
conn.commit()
|
|
|
|
# Initialize statistics table with single row
|
|
db = SessionLocal()
|
|
try:
|
|
stats = db.query(Statistics).first()
|
|
if not stats:
|
|
stats = Statistics(
|
|
id=1,
|
|
total_events=0,
|
|
total_tools_used=0,
|
|
total_agents=0,
|
|
total_sessions=0,
|
|
last_updated=0.0,
|
|
)
|
|
db.add(stats)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def get_db():
|
|
"""Dependency for getting database session."""
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|