Function apply_migrations

plugins/codex-security/scripts/workbench_db.py:416–445

(connection: sqlite3.Connection)

def apply_migrations(connection: sqlite3.Connection) -> None:
    connection.commit()
    connection.execute("BEGIN IMMEDIATE")
    try:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS schema_migrations (
                version INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                applied_at TEXT NOT NULL
            )
            """
        )
        normalize_pre_release_migrations(connection)
        applied = {
            row["version"] for row in connection.execute("SELECT version FROM schema_migrations")
        }
        for version, name, sql in MIGRATIONS:
            if version in applied:
                continue
            for statement in sql_statements(sql):
                connection.execute(statement)
            connection.execute(
                "INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)",
                (version, name, now()),
            )
        connection.commit()
    except BaseException:
        connection.rollback()
        raise


def normalize_pre_release_migrations(connection: sqlite3.Connection) -> None: