patrickandClaude Sonnet 5.5
347f55cf7c
Includes atomic migrations, per-rule reminder semantics, first-run helper script and the note on the undecided UI chat idea. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
import sqlite3
|
|
from typing import Sequence
|
|
|
|
|
|
class MigrationError(Exception):
|
|
pass
|
|
|
|
|
|
def split_statements(script: str) -> list[str]:
|
|
statements: list[str] = []
|
|
buffer = ""
|
|
for line in script.splitlines(keepends=True):
|
|
buffer += line
|
|
if sqlite3.complete_statement(buffer):
|
|
statements.append(buffer.strip())
|
|
buffer = ""
|
|
if buffer.strip():
|
|
statements.append(buffer.strip())
|
|
return statements
|
|
|
|
|
|
def apply_migration(conn: sqlite3.Connection, index: int, script: str) -> None:
|
|
try:
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
for statement in split_statements(script):
|
|
conn.execute(statement)
|
|
conn.execute(f"PRAGMA user_version={index + 1}")
|
|
conn.execute("COMMIT")
|
|
except Exception as exc:
|
|
try:
|
|
conn.execute("ROLLBACK")
|
|
except sqlite3.Error:
|
|
pass
|
|
raise MigrationError(f"Migration {index} fehlgeschlagen und zurückgerollt: {exc}") from exc
|
|
|
|
|
|
def run_migrations(conn: sqlite3.Connection, migrations: Sequence[str]) -> None:
|
|
version = conn.execute("PRAGMA user_version").fetchone()[0]
|
|
# Im Autocommit-Modus, damit BEGIN/COMMIT ausschließlich hier gesetzt werden (executescript würde implizit committen).
|
|
previous = conn.isolation_level
|
|
conn.isolation_level = None
|
|
try:
|
|
for index in range(version, len(migrations)):
|
|
apply_migration(conn, index, migrations[index])
|
|
finally:
|
|
conn.isolation_level = previous
|