summaryrefslogtreecommitdiff
path: root/database/database.py
diff options
context:
space:
mode:
Diffstat (limited to 'database/database.py')
-rw-r--r--database/database.py73
1 files changed, 0 insertions, 73 deletions
diff --git a/database/database.py b/database/database.py
deleted file mode 100644
index 7119c7e..0000000
--- a/database/database.py
+++ /dev/null
@@ -1,73 +0,0 @@
-import sqlite3
-from datetime import UTC, datetime
-from pathlib import Path
-
-
-class Database:
- def __init__(self, path: Path) -> None:
- self.path = path
- self.migrations_dir = Path(__file__).parent / "migrations"
-
- def connect(self) -> sqlite3.Connection:
- return sqlite3.connect(self.path)
-
- def migrate(self) -> None:
- with self.connect() as connection:
- connection.execute(
- """
- CREATE TABLE IF NOT EXISTS schema_migrations (
- version INTEGER PRIMARY KEY,
- applied_at TEXT NOT NULL
- )
- """
- )
-
- applied_versions = {
- row[0]
- for row in connection.execute("SELECT version FROM schema_migrations")
- }
-
- for migration_version, path in self._get_migrations():
- if migration_version in applied_versions:
- continue
-
- sql = path.read_text(encoding="utf-8")
-
- connection.execute("BEGIN")
-
- try:
- connection.execute(sql)
-
- applied_at = datetime.now(UTC).isoformat().replace("+00:00", "Z")
-
- connection.execute(
- """
- INSERT INTO schema_migrations (version, applied_at)
- VALUES (?, ?)
- """,
- (migration_version, applied_at),
- )
-
- connection.commit()
- except Exception:
- connection.rollback()
- raise
-
- def _get_migrations(self) -> list[tuple[int, Path]]:
- migrations = []
-
- for path in self.migrations_dir.glob("*.sql"):
- version_text, _, _ = path.stem.partition("_")
-
- if not version_text.isdigit():
- continue
-
- migrations.append((int(version_text), path))
-
- migrations.sort()
-
- versions = [version for version, _ in migrations]
- if len(versions) != len(set(versions)):
- raise ValueError("Duplicate migration version")
-
- return migrations