153 lines
5.3 KiB
Python
153 lines
5.3 KiB
Python
"""SQLite mapping database for SVN↔Git revision mapping."""
|
|
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
class MappingDB:
|
|
def __init__(self, path: Path):
|
|
self.path = path
|
|
self.conn = sqlite3.connect(str(path))
|
|
self.conn.row_factory = sqlite3.Row
|
|
self.conn.execute("PRAGMA journal_mode=WAL")
|
|
self.conn.execute("PRAGMA synchronous=NORMAL")
|
|
self._init_schema()
|
|
|
|
def _init_schema(self):
|
|
self.conn.executescript("""
|
|
CREATE TABLE IF NOT EXISTS schema_version (
|
|
version INTEGER PRIMARY KEY
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS svn_to_git (
|
|
svn_revision INTEGER NOT NULL,
|
|
git_commit_hash TEXT NOT NULL,
|
|
svn_branch TEXT NOT NULL,
|
|
git_ref TEXT NOT NULL,
|
|
synced_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
|
|
source TEXT NOT NULL DEFAULT 'svn'
|
|
CHECK (source IN ('svn', 'git')),
|
|
UNIQUE(svn_revision, svn_branch)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_svn_to_git_hash
|
|
ON svn_to_git(git_commit_hash);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_svn_to_git_svn
|
|
ON svn_to_git(svn_revision);
|
|
|
|
CREATE TABLE IF NOT EXISTS sync_state (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
);
|
|
|
|
INSERT OR IGNORE INTO schema_version (version) VALUES (1);
|
|
""")
|
|
self.conn.commit()
|
|
|
|
def close(self):
|
|
self.conn.close()
|
|
|
|
# --- Mapping CRUD ---
|
|
|
|
def record_mapping(
|
|
self,
|
|
svn_revision: int,
|
|
git_commit_hash: str,
|
|
svn_branch: str,
|
|
git_ref: str,
|
|
source: str = "svn",
|
|
):
|
|
self.conn.execute(
|
|
"""INSERT OR REPLACE INTO svn_to_git
|
|
(svn_revision, git_commit_hash, svn_branch, git_ref, source)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
(svn_revision, git_commit_hash, svn_branch, git_ref, source),
|
|
)
|
|
self.conn.commit()
|
|
|
|
def get_git_hash(self, svn_revision: int, svn_branch: str) -> Optional[str]:
|
|
row = self.conn.execute(
|
|
"SELECT git_commit_hash FROM svn_to_git WHERE svn_revision=? AND svn_branch=?",
|
|
(svn_revision, svn_branch),
|
|
).fetchone()
|
|
return row[0] if row else None
|
|
|
|
def get_svn_revision(self, git_commit_hash: str) -> Optional[int]:
|
|
row = self.conn.execute(
|
|
"SELECT svn_revision FROM svn_to_git WHERE git_commit_hash=?",
|
|
(git_commit_hash,),
|
|
).fetchone()
|
|
return row[0] if row else None
|
|
|
|
def get_last_svn_revision(self, svn_branch: str = "trunk") -> Optional[int]:
|
|
row = self.conn.execute(
|
|
"SELECT MAX(svn_revision) FROM svn_to_git WHERE svn_branch=?",
|
|
(svn_branch,),
|
|
).fetchone()
|
|
return row[0] if row and row[0] is not None else None
|
|
|
|
def mapping_exists(self, svn_revision: int, svn_branch: str) -> bool:
|
|
row = self.conn.execute(
|
|
"SELECT 1 FROM svn_to_git WHERE svn_revision=? AND svn_branch=?",
|
|
(svn_revision, svn_branch),
|
|
).fetchone()
|
|
return row is not None
|
|
|
|
def get_all_mappings(self):
|
|
return self.conn.execute(
|
|
"SELECT svn_revision, git_commit_hash, svn_branch, git_ref, source "
|
|
"FROM svn_to_git ORDER BY svn_revision"
|
|
).fetchall()
|
|
|
|
def ref_exists(self, git_ref: str) -> bool:
|
|
# Also check the git-svn remote variant if this is a canonical ref
|
|
variants = [git_ref]
|
|
if git_ref.startswith("refs/tags/"):
|
|
variants.append("refs/remotes/origin/tags/" + git_ref[10:])
|
|
elif git_ref == "refs/heads/master":
|
|
variants.append("refs/remotes/origin/trunk")
|
|
elif git_ref.startswith("refs/heads/"):
|
|
variants.append("refs/remotes/origin/" + git_ref[11:])
|
|
row = self.conn.execute(
|
|
"SELECT 1 FROM svn_to_git WHERE git_ref IN ({})".format(
|
|
",".join("?" * len(variants))
|
|
), variants
|
|
).fetchone()
|
|
return row is not None
|
|
|
|
def get_mapping_by_git_hash(self, git_hash: str) -> Optional[dict]:
|
|
row = self.conn.execute(
|
|
"SELECT svn_revision, svn_branch, git_ref, source "
|
|
"FROM svn_to_git WHERE git_commit_hash=? LIMIT 1",
|
|
(git_hash,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_max_revision_across_all_branches(self) -> Optional[int]:
|
|
row = self.conn.execute(
|
|
"SELECT MAX(svn_revision) FROM svn_to_git",
|
|
).fetchone()
|
|
return row[0] if row and row[0] is not None else None
|
|
|
|
def mapping_count(self) -> int:
|
|
row = self.conn.execute("SELECT COUNT(*) FROM svn_to_git").fetchone()
|
|
return row[0] if row else 0
|
|
|
|
# --- Sync State ---
|
|
|
|
def set_state(self, key: str, value: str):
|
|
self.conn.execute(
|
|
"INSERT OR REPLACE INTO sync_state (key, value) VALUES (?, ?)",
|
|
(key, value),
|
|
)
|
|
self.conn.commit()
|
|
|
|
def get_state(self, key: str) -> Optional[str]:
|
|
row = self.conn.execute(
|
|
"SELECT value FROM sync_state WHERE key=?", (key,)
|
|
).fetchone()
|
|
return row[0] if row else None
|