"""Database migrations for Coach AI."""
import aiosqlite
async def migrate_database(db: aiosqlite.Connection) -> None:
"""Run all necessary database migrations.
This function checks for and applies missing columns/tables
to upgrade the database schema without losing data.
"""
# Check current schema version
cursor = await db.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='schema_version'"
)
version_table_exists = await cursor.fetchone()
if not version_table_exists:
# Create version tracking table
await db.execute(
"""
CREATE TABLE schema_version (
version INTEGER PRIMARY KEY,
applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
await db.execute("INSERT INTO schema_version (version) VALUES (0)")
await db.commit()
# Get current version
cursor = await db.execute("SELECT MAX(version) FROM schema_version")
row = await cursor.fetchone()
current_version = row[0] if row else 0
# Apply migrations in order
if current_version < 1:
await _migration_1_add_task_tracking(db)
await db.execute("INSERT INTO schema_version (version) VALUES (1)")
await db.commit()
if current_version < 2:
await _migration_2_temporal_planning(db)
await db.execute("INSERT INTO schema_version (version) VALUES (2)")
await db.commit()
if current_version < 3:
await _migration_3_add_quick_column(db)
await db.execute("INSERT INTO schema_version (version) VALUES (3)")
await db.commit()
async def _migration_1_add_task_tracking(db: aiosqlite.Connection) -> None:
"""Migration 1: Add task tracking columns to todos table.
Adds:
- skipped_count: Track how many times task was moved forward
- time_estimate: Estimated time in minutes for task completion
- last_scheduled: Last date this task was scheduled in a daily note
"""
# Check if columns already exist (safe migration)
cursor = await db.execute("PRAGMA table_info(todos)")
columns = {row[1] for row in await cursor.fetchall()}
if "skipped_count" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN skipped_count INTEGER DEFAULT 0")
if "time_estimate" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN time_estimate INTEGER")
if "last_scheduled" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN last_scheduled DATE")
await db.commit()
async def _migration_2_temporal_planning(db: aiosqlite.Connection) -> None:
"""Migration 2: Add temporal planning and executive dysfunction support fields.
Adds to todos table:
- timeframe: When to work on task (this_week, next_sprint, this_month, this_quarter, someday)
- energy_required: Energy level needed (high, medium, low)
- theme_tag: Type of work (sprint_work, strategic, admin, learning)
- task_context: Free text for progress notes, blockers, why doing this
- blocked_by: What's blocking this task
- linked_goal_id: Link to goals table
- parent_todo_id: For subtasks (links to parent todo)
- task_order: Order of subtasks within parent
Creates new tables:
- week_themes: Store weekly work theme planning
- weekly_reviews: Historical weekly review data
"""
# Check existing columns
cursor = await db.execute("PRAGMA table_info(todos)")
columns = {row[1] for row in await cursor.fetchall()}
# Add new columns to todos table
if "timeframe" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN timeframe TEXT DEFAULT NULL")
if "energy_required" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN energy_required TEXT DEFAULT 'medium'")
if "theme_tag" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN theme_tag TEXT DEFAULT NULL")
if "task_context" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN task_context TEXT DEFAULT NULL")
if "blocked_by" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN blocked_by TEXT DEFAULT NULL")
if "linked_goal_id" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN linked_goal_id INTEGER DEFAULT NULL")
if "parent_todo_id" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN parent_todo_id INTEGER DEFAULT NULL")
if "task_order" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN task_order INTEGER DEFAULT 0")
# Create indices for performance
await db.execute("CREATE INDEX IF NOT EXISTS idx_todos_timeframe ON todos(timeframe)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_todos_theme ON todos(theme_tag)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_todos_parent ON todos(parent_todo_id)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_todos_linked_goal ON todos(linked_goal_id)")
# Create week_themes table
await db.execute("""
CREATE TABLE IF NOT EXISTS week_themes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
week_start_date DATE NOT NULL UNIQUE,
theme_json TEXT,
focus_items TEXT,
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Create weekly_reviews table
await db.execute("""
CREATE TABLE IF NOT EXISTS weekly_reviews (
id INTEGER PRIMARY KEY AUTOINCREMENT,
week_start_date DATE NOT NULL UNIQUE,
completed_count INTEGER DEFAULT 0,
strategic_progress TEXT,
sprint_progress TEXT,
reflection_notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
await db.commit()
async def _migration_3_add_quick_column(db: aiosqlite.Connection) -> None:
"""Migration 3: Add quick column to todos table.
Adds:
- quick: Boolean flag for marking quick wins (low activation energy tasks)
"""
# Check if column already exists (safe migration)
cursor = await db.execute("PRAGMA table_info(todos)")
columns = {row[1] for row in await cursor.fetchall()}
if "quick" not in columns:
await db.execute("ALTER TABLE todos ADD COLUMN quick BOOLEAN DEFAULT FALSE")
await db.commit()
