63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
|
|
|
|
class Database:
|
|
def __init__(self, path: Path, migrations_path: Path) -> None:
|
|
self.path = path
|
|
self.migrations_path = migrations_path
|
|
|
|
@contextmanager
|
|
def connect(self) -> Iterator[sqlite3.Connection]:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
connection = sqlite3.connect(self.path, timeout=10, check_same_thread=False)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA foreign_keys = ON")
|
|
connection.execute("PRAGMA journal_mode = WAL")
|
|
connection.execute("PRAGMA busy_timeout = 10000")
|
|
try:
|
|
yield connection
|
|
finally:
|
|
connection.close()
|
|
|
|
def migrate(self) -> None:
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
version INTEGER PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
applied_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
connection.commit()
|
|
applied = {
|
|
row["version"]
|
|
for row in connection.execute("SELECT version FROM schema_migrations").fetchall()
|
|
}
|
|
for path in sorted(self.migrations_path.glob("*.sql")):
|
|
version = int(path.name.split("_", 1)[0])
|
|
if version in applied:
|
|
continue
|
|
sql = path.read_text(encoding="utf-8")
|
|
name = path.stem.replace("'", "''")
|
|
connection.executescript(
|
|
"BEGIN IMMEDIATE;\n"
|
|
f"{sql}\n"
|
|
"INSERT INTO schema_migrations(version, name, applied_at) "
|
|
f"VALUES ({version}, '{name}', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));\n"
|
|
"COMMIT;"
|
|
)
|
|
|
|
def health_check(self) -> bool:
|
|
try:
|
|
with self.connect() as connection:
|
|
return connection.execute("SELECT 1").fetchone()[0] == 1
|
|
except sqlite3.Error:
|
|
return False
|