v0_3_roadmap.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/migrations/v0_3_roadmap.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_schema_migration.py"
"""
Runtime migration for the roadmap v0.3 registry extension.
 
Purpose
-------
Apply and roll back the activated roadmap v0.3 SQL extension.
 
Responsibilities
----------------
- Load the canonical REGISTRY_SCHEMA_V0_3.sql file.
- Create roadmap tables idempotently and set PRAGMA user_version to 3.
- Drop roadmap tables during rollback for migration tests and recovery.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - migration lifecycle
    L3 - schema version decisions
    L4 - table existence checks
 
Contracts
---------
- The SQL file is the single source of truth for created objects.
- The module does not import wall-clock APIs; callers provide timestamps.
 
Dependencies
------------
- sqlite3 (migration connection)
- oc_diagnostics (_dbg structured logger)
"""
from __future__ import annotations
 
from pathlib import Path
import sqlite3
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
ROADMAP_SCHEMA_VERSION = 3
ROADMAP_SCHEMA_DESCRIPTION = "roadmap REGISTRY_SCHEMA_V0_3"
 
ROADMAP_TABLES = (
    "roadmap_eras",
    "roadmap_versions",
    "roadmap_phases",
    "roadmap_tracks",
    "roadmap_items_v3",
    "roadmap_item_tracks",
    "roadmap_item_dependencies",
    "roadmap_item_events",
    "roadmap_plan_links",
    "roadmap_render_sections",
)
 
 
def schema_sql_path() -> Path:
    """Return the canonical v0.3 roadmap SQL extension path."""
 
    return (
        Path(__file__).resolve().parents[2]
        / "docs"
        / "architecture"
        / "orchestration"
        / "REGISTRY_SCHEMA_V0_3.sql"
    )
 
 
def load_schema_sql() -> str:
    """Load the canonical v0.3 roadmap SQL extension."""
 
    return schema_sql_path().read_text(encoding="utf-8")
 
 
def current_user_version(conn: sqlite3.Connection) -> int:
    """Return SQLite's integer schema user version."""
 
    return int(conn.execute("PRAGMA user_version").fetchone()[0])
 
 
def apply(conn: sqlite3.Connection) -> None:
    """Apply the roadmap v0.3 schema extension idempotently."""
 
    if current_user_version(conn) >= ROADMAP_SCHEMA_VERSION and tables_exist(conn):
        return
    conn.executescript(load_schema_sql())
 
 
def rollback(conn: sqlite3.Connection) -> None:
    """Drop roadmap v0.3 tables and reset SQLite user_version to 2."""
 
    for name in reversed(ROADMAP_TABLES):
        conn.execute(f"DROP TABLE IF EXISTS {name}")
    # OCT-LINT: disable=no_hardcoded_secrets reason="Schema object names are migration constants, not secrets"
    conn.execute("DROP TRIGGER IF EXISTS roadmap_item_events_no_update")
    conn.execute("DROP TRIGGER IF EXISTS roadmap_item_events_no_delete")
    conn.execute("PRAGMA user_version = 2")
    conn.execute(
        "DELETE FROM schema_migrations WHERE version = ?",
        (ROADMAP_SCHEMA_VERSION,),
    )
 
 
def tables_exist(conn: sqlite3.Connection) -> bool:
    """Return True when all roadmap v0.3 tables exist."""
 
    rows = conn.execute(
        "SELECT name FROM sqlite_master WHERE type = 'table'"
    ).fetchall()
    existing = {str(row[0]) for row in rows}
    return set(ROADMAP_TABLES) <= existing
 
 
def table_names() -> tuple[str, ...]:
    """Return roadmap v0.3 table names in dependency order."""
 
    return ROADMAP_TABLES