32 lines
1.5 KiB
Python
32 lines
1.5 KiB
Python
|
"""Incremental schema migration register for SQLite single-host installations.
|
||
|
|
|
||
|
|
Legacy tables are created by their existing setup functions. This new ledger
|
||
|
|
tracks newly introduced migrations and verifies their SQL checksums. Do not
|
||
|
|
mistake this for a completed migration of all legacy PostgreSQL structures.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
import hashlib
|
||
|
|
from datetime import datetime,timezone
|
||
|
|
|
||
|
|
# Each migration has separately executable (single statement) DDL entries.
|
||
|
|
# Never rewrite an already applied migration: add a new sequence instead.
|
||
|
|
MIGRATIONS=(
|
||
|
|
('20261022_001_p1_limits',(
|
||
|
|
'CREATE TABLE IF NOT EXISTS platform_limit_overrides (name TEXT PRIMARY KEY, value INTEGER NOT NULL CHECK (value > 0), updated_at TEXT NOT NULL)',
|
||
|
|
)),
|
||
|
|
)
|
||
|
|
|
||
|
|
def ensure(conn):
|
||
|
|
conn.execute('''CREATE TABLE IF NOT EXISTS schema_migrations (
|
||
|
|
version TEXT PRIMARY KEY, checksum TEXT NOT NULL, applied_at TEXT NOT NULL)''')
|
||
|
|
for version,statements in MIGRATIONS:
|
||
|
|
checksum=hashlib.sha256('\n'.join(statements).encode()).hexdigest()
|
||
|
|
row=conn.execute('SELECT checksum FROM schema_migrations WHERE version=?',(version,)).fetchone()
|
||
|
|
if row:
|
||
|
|
if row[0]!=checksum:raise RuntimeError('Schema migration checksum mismatch: '+version)
|
||
|
|
continue
|
||
|
|
# The caller owns a single transaction. DDL and ledger entry commit together.
|
||
|
|
for sql in statements:conn.execute(sql)
|
||
|
|
conn.execute('INSERT INTO schema_migrations VALUES (?,?,?)',(
|
||
|
|
version,checksum,datetime.now(timezone.utc).isoformat(timespec='seconds')))
|