blob: 22e5b48e9fd3fc12fcfd3bf3d68a8a682533d7d7 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
|
"""Database connection and migration management for senia.db.
Single shared database for all senia features. Tables for new features are added
via new migration files; the runner detects and applies them automatically.
"""
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
DB_PATH = Path("/etc/senia/data/senia.db")
MIGRATIONS_DIR = Path(__file__).parent / "migrations"
def get_conn() -> sqlite3.Connection:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
return conn
def _ensure_migrations_table(conn: sqlite3.Connection) -> None:
conn.execute("""
CREATE TABLE IF NOT EXISTS schema_migrations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
filename TEXT NOT NULL UNIQUE,
applied_at TEXT NOT NULL
)
""")
def run_migrations() -> None:
"""Apply any pending migrations from migrations/ in order."""
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
with get_conn() as conn:
_ensure_migrations_table(conn)
applied = {r["filename"] for r in
conn.execute("SELECT filename FROM schema_migrations").fetchall()}
for f in sorted(MIGRATIONS_DIR.glob("*.sql")):
if f.name not in applied:
conn.executescript(f.read_text())
conn.execute(
"INSERT INTO schema_migrations (filename, applied_at) VALUES (?, ?)",
(f.name, datetime.now(timezone.utc).isoformat()),
)
print(f"applied migration: {f.name}")
|
