LeoVasanko/paskia
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Migrate command: merge legacy and current databases into existing paskia.kantadb

Browse Source 
- paskia migrate accepts an rp-id, a legacy *.paskiadb path, or a
  current-format *.kantadb path; with an existing target database the
  incoming data is merged (uuid-keyed records make conflicts a non-issue,
  domains merge per rp-id with a union of origins)
- Migration transactions are labeled migrate:cli:{rp-id} (slash-joined
  for multi-domain sources) instead of 'bootstrap'
This commit is contained in:
LeoVasanko
2026-09-09 17:23:27 +00:00
parent baa7e47187
commit ae1928241e
3 changed files with 286 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paskia/__main__.py
+13 -7
View File
@@ -178,9 +178,11 @@ def cmd_init(args: argparse.Namespace) -> None:
def cmd_migrate(args: argparse.Namespace) -> None:
"""Convert a legacy <rp-id>.paskiadb database to paskia.kantadb."""
rp_id = legacy.migrate_legacy_database(args.rp_id)
print(f"✅ Converted legacy database to {db_file_path()} (domain: {rp_id})")
"""Convert or merge a legacy/current database into paskia.kantadb."""
merging = db_file_path().exists()
rp_ids = legacy.migrate_database(args.source)
action = "Merged into existing" if merging else "Converted to"
print(f"✅ {action} {db_file_path()} (domains: {', '.join(rp_ids)})")
def cmd_serve(args: argparse.Namespace) -> None:
@@ -263,14 +265,18 @@ def main():
migrate_parser = argparse.ArgumentParser(
prog="paskia migrate",
description="Convert a legacy <rp-id>.paskiadb database to paskia.kantadb",
description="Convert a legacy <rp-id>.paskiadb database to paskia.kantadb, "
"or merge a legacy database / another paskia.kantadb into an existing one",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
migrate_parser.add_argument(
"rp_id",
"source",
nargs="?",
help="rp-id of the legacy database to convert, selecting "
"<rp-id>.paskiadb when several legacy candidates exist.",
help="rp-id of the legacy database to convert, or path to a legacy "
"<rp-id>.paskiadb directory/file or a current-format paskia.kantadb "
"file. When paskia.kantadb already exists, the source data is merged "
"into it. Without an argument, a single legacy *.paskiadb candidate "
"in the current directory is selected automatically.",
)
argv = sys.argv[1:]
paskia/db/legacy.py
+181 -56
View File
@@ -1,14 +1,15 @@
"""Legacy database format reader and converter.
"""Legacy database format reader, converter and database merging.
Retains the msgspec structs used by the old ``<rp-id>.paskiadb/main.db``
format so existing databases can be opened and converted to the combined
``paskia.kantadb`` format. Only the structs whose shape differs from the
current schema are redefined here; unchanged structs are imported from
``paskia.db.structs``.
``paskia.kantadb`` format, and implements the merge of incoming data
(legacy or current format) into an existing ``paskia.kantadb``. Only the
structs whose shape differs from the current schema are redefined here;
unchanged structs are imported from ``paskia.db.structs``.
Assumes the on-disk records are in the latest legacy format (schema
migrations were discarded together with the old format). This module will
be deleted once legacy conversion is no longer supported.
migrations were discarded together with the old format). The legacy
structs will be deleted once legacy conversion is no longer supported.
"""
from __future__ import annotations
@@ -101,16 +102,23 @@ def _read_legacy(path: Path) -> LegacyDB:
return asyncio.run(_read())
def convert_legacy_database(src: Path, dst: Path) -> Config:
"""Convert a legacy main.db file into the combined kantadb format.
def _read_kantadb(path: Path) -> DB:
"""Open a current-format database read-only and return its contents."""
kanta = Kanta(str(path), DB())
Reads the legacy database at ``src`` and writes a fresh database at
``dst``. All credentials and sessions are stamped with the legacy
database's rp-id; the OIDC provider carries over as-is (it is
instance-global).
Returns the converted (new-format) configuration.
async def _read() -> DB:
await kanta.open(readonly=True)
return kanta.data
return asyncio.run(_read())
def _legacy_to_db(old: LegacyDB) -> DB:
"""Convert legacy database contents to the combined kantadb format.
All credentials and sessions are stamped with the legacy database's
rp-id; the OIDC provider carries over as-is (it is instance-global).
"""
old = _read_legacy(src)
rp_id = old.config.rp_id
from paskia.domains import origin_key # noqa: PLC0415 (import cycle)
@@ -169,28 +177,94 @@ def convert_legacy_database(src: Path, dst: Path) -> Config:
reset_tokens=old.reset_tokens,
oidc=old.oidc,
)
return converted
def _migration_label(incoming: DB) -> str:
"""Transaction label for a migration; multiple rp-ids join with slashes."""
return f"migrate:cli:{'/'.join(incoming.config.domains)}"
def _write_fresh(data: DB, dst: Path, label: str) -> None:
"""Write a fresh database at ``dst`` with the given contents."""
new_db = DB()
kanta = Kanta(str(dst), new_db)
@kanta.bootstrap
def _seed(data: DB) -> None:
data.config = converted.config
data.permissions = converted.permissions
data.orgs = converted.orgs
data.roles = converted.roles
data.users = converted.users
data.credentials = converted.credentials
data.sessions = converted.sessions
data.reset_tokens = converted.reset_tokens
data.oidc = converted.oidc
@kanta.bootstrap(action=label)
def _seed(target: DB) -> None:
target.config = data.config
target.permissions = data.permissions
target.orgs = data.orgs
target.roles = data.roles
target.users = data.users
target.credentials = data.credentials
target.sessions = data.sessions
target.reset_tokens = data.reset_tokens
target.oidc = data.oidc
async def _write() -> None:
async with kanta:
pass
asyncio.run(_write())
return new_config
def convert_legacy_database(src: Path, dst: Path) -> Config:
"""Convert a legacy main.db file into the combined kantadb format.
Reads the legacy database at ``src`` and writes a fresh database at
``dst``. Returns the converted (new-format) configuration.
"""
converted = _legacy_to_db(_read_legacy(src))
_write_fresh(converted, dst, _migration_label(converted))
return converted.config
def _merge_data(data: DB, incoming: DB) -> None:
"""Merge ``incoming`` contents into the live ``data`` object.
Records are uuid-keyed (or hash-keyed for sessions/reset tokens), so
identical keys denote the same item: existing entries win, new entries
are added. Domains merge per rp-id with a union of allowed origins;
the existing instance's listen endpoints and OIDC signing key win.
"""
for rp_id, domain in incoming.config.domains.items():
existing = data.config.domains.get(rp_id)
if existing is None:
data.config.domains[rp_id] = domain
continue
for origin, entry in domain.origins.items():
existing.origins.setdefault(origin, entry)
if existing.rp_name is None:
existing.rp_name = domain.rp_name
for bucket in (
"permissions",
"orgs",
"roles",
"users",
"credentials",
"sessions",
"reset_tokens",
):
target_map = getattr(data, bucket)
for key, value in getattr(incoming, bucket).items():
target_map.setdefault(key, value)
for uuid, client in incoming.oidc.clients.items():
data.oidc.clients.setdefault(uuid, client)
if data.oidc.key is None:
data.oidc.key = incoming.oidc.key
def merge_database(dst: Path, incoming: DB) -> None:
"""Merge ``incoming`` contents into the existing database at ``dst``."""
kanta = Kanta(str(dst), DB())
async def _merge() -> None:
async with kanta:
with kanta.transaction(_migration_label(incoming)):
_merge_data(kanta.data, incoming)
asyncio.run(_merge())
def find_legacy_databases(cwd: Path | None = None) -> list[Path]:
@@ -211,50 +285,101 @@ def find_legacy_databases(cwd: Path | None = None) -> list[Path]:
return candidates
def migrate_legacy_database(rp_id: str | None = None) -> str:
"""Convert a legacy database to ``paskia.kantadb``.
def _resolve_source(source: str | None) -> tuple[Path, bool, Path, Path | None]:
"""Resolve the migrate source.
With ``rp_id``, selects the ``<rp-id>.paskiadb`` candidate by name;
without it, exactly one candidate must exist. Returns the migrated
domain's rp-id. The converted legacy directory/file is renamed aside
to ``<name>.converted-bak`` rather than deleted.
``source`` may be an rp-id (selecting ``<rp-id>.paskiadb`` in the
current directory), a path to a legacy ``*.paskiadb`` directory or
file, or a path to a current-format ``*.kantadb`` file. Without
``source``, exactly one legacy candidate must exist in the current
directory.
Raises SystemExit when ``paskia.kantadb`` already exists, when no
candidate matches, or when several candidates exist and no ``rp_id``
was given to select one.
Returns ``(db_file, is_legacy, users_dir, rename_target)`` where
``users_dir`` holds auxiliary user files (avatars) and
``rename_target`` is the legacy directory/file to rename aside after
a successful migration (None for current-format sources).
"""
target = db_file_path()
if target.exists():
raise SystemExit(f"Database {target} already exists — nothing to migrate.")
candidates = find_legacy_databases()
if rp_id is not None:
name = f"{rp_id}.paskiadb"
matches = [c for c in candidates if c.name == name]
def legacy(src: Path) -> tuple[Path, bool, Path, Path]:
return (
src / "main.db" if src.is_dir() else src,
True,
src / "users" if src.is_dir() else src.parent / "users",
src,
)
if source is not None:
path = Path(source)
if path.is_dir():
if (path / "main.db").is_file():
return legacy(path)
raise SystemExit(f"No legacy main.db found in directory {path}.")
if path.is_file():
if path.suffix == ".paskiadb":
return legacy(path)
return path, False, path.parent / "paskia.data" / "users", None
# Not a path: treat as rp-id selecting a legacy candidate by name
name = f"{source}.paskiadb"
matches = [c for c in find_legacy_databases() if c.name == name]
if not matches:
found = ", ".join(str(c) for c in candidates) or "none"
found = ", ".join(str(c) for c in find_legacy_databases()) or "none"
raise SystemExit(
f"No legacy database {name} in this directory (candidates: {found})."
)
src = matches[0]
elif not candidates:
return legacy(matches[0])
candidates = find_legacy_databases()
if not candidates:
raise SystemExit("No legacy *.paskiadb database found — nothing to migrate.")
elif len(candidates) > 1:
if len(candidates) > 1:
names = ", ".join(str(c) for c in candidates)
raise SystemExit(
f"Multiple legacy databases found ({names}) — select one with "
"'paskia migrate <rp-id>'."
)
else:
src = candidates[0]
legacy_file = src / "main.db" if src.is_dir() else src
config = convert_legacy_database(legacy_file, target)
return legacy(candidates[0])
# Move persisted user files (avatars) to the new data root
legacy_users = src / "users" if src.is_dir() else None
if legacy_users is not None and legacy_users.is_dir():
def _move_user_files(src_users: Path) -> None:
"""Move persisted user files (avatars) to the new data root."""
if not src_users.is_dir():
return
target_users = users_root_path(create_root=True)
for child in legacy_users.iterdir():
for child in src_users.iterdir():
if (target_users / child.name).exists():
continue
shutil.move(str(child), str(target_users / child.name))
shutil.move(str(src), str(src.with_name(src.name + ".converted-bak")))
return next(iter(config.domains))
def migrate_database(source: str | None = None) -> list[str]:
"""Convert or merge a database into ``paskia.kantadb``.
The source may be a legacy ``<rp-id>.paskiadb`` database (selected by
rp-id or path) or a current-format ``*.kantadb`` file given by path.
When ``paskia.kantadb`` already exists, the incoming data is merged
into it (uuid-keyed records make conflicts a non-issue); otherwise a
fresh database is written. Returns the migrated domains' rp-ids. A
migrated legacy source is renamed aside to ``<name>.converted-bak``
rather than deleted; a merged kantadb source is left in place.
"""
target = db_file_path()
db_file, is_legacy, users_dir, rename_target = _resolve_source(source)
if db_file.resolve() == target.resolve():
raise SystemExit(f"{db_file} is the active database — nothing to migrate.")
incoming = (
_legacy_to_db(_read_legacy(db_file)) if is_legacy else _read_kantadb(db_file)
)
rp_ids = list(incoming.config.domains)
if target.exists():
merge_database(target, incoming)
else:
_write_fresh(incoming, target, _migration_label(incoming))
_move_user_files(users_dir)
if rename_target is not None and rename_target.exists():
shutil.move(
str(rename_target),
str(rename_target.with_name(rename_target.name + ".converted-bak")),
)
return rp_ids
tests/test_cli.py
+90 -5
View File
@@ -2,7 +2,8 @@
The CLI is split into ``paskia init`` (create the combined paskia.kantadb
with the initial domain(s)), ``paskia migrate`` (convert a legacy
``<rp-id>.paskiadb`` database), and bare ``paskia`` (serve the stored
``<rp-id>.paskiadb`` database, or merge a legacy/current database into an
existing paskia.kantadb), and bare ``paskia`` (serve the stored
domains; never migrates).
"""
@@ -20,7 +21,7 @@ from kanta import Kanta
from paskia.__main__ import _load_stored_config, main
from paskia.db import legacy
from paskia.db.structs import DB, Config
from paskia.db.structs import DB, Config, DomainConfig
from paskia.util.runtime import ServeConfig, clear_cache
@@ -204,6 +205,8 @@ def test_migrate_converts_legacy_database(run_cli, tmp_path):
config = stored_config(tmp_path)
assert list(config.domains) == ["example.com"]
assert config.domains["example.com"].rp_name == "Legacy Name"
# Migration transaction is labeled with the migrated rp-id
assert b"migrate:cli:example.com" in (tmp_path / "paskia.kantadb").read_bytes()
# Legacy directory renamed aside, user files moved over
assert not src_dir.exists()
assert (tmp_path / "example.com.paskiadb.converted-bak").is_dir()
@@ -242,11 +245,93 @@ def test_migrate_unknown_rp_id(run_cli, tmp_path):
run_cli("migrate", "nope.com")
def test_migrate_refuses_existing_database(run_cli):
run_cli("init")
with pytest.raises(SystemExit, match="already exists"):
def test_migrate_merges_legacy_into_existing_database(run_cli, tmp_path):
"""An existing paskia.kantadb is not refused — data is merged in."""
run_cli("init", "company.com", "Company")
write_legacy_db(tmp_path, legacy.LegacyConfig(rp_id="example.com", rp_name="Ex"))
run_cli("migrate")
config = stored_config(tmp_path)
assert list(config.domains) == ["company.com", "example.com"]
assert config.domains["example.com"].rp_name == "Ex"
assert (tmp_path / "example.com.paskiadb.converted-bak").is_dir()
def write_kantadb(root: Path, domains: dict, name: str = "paskia.kantadb") -> Path:
"""Create a current-format database file with the given config domains."""
db_file = root / name
config = Config(
domains={rp_id: DomainConfig(rp_name=name_) for rp_id, name_ in domains.items()}
)
async def _write() -> None:
new_db = DB()
kanta = Kanta(str(db_file), new_db)
@kanta.bootstrap
def _seed(data: DB) -> None:
data.config = config
async with kanta:
pass
asyncio.run(_write())
return db_file
def test_migrate_merges_kantadb_into_existing_database(run_cli, tmp_path):
run_cli("init", "company.com", "Company")
src = write_kantadb(tmp_path, {"other.com": "Other"}, name="other.kantadb")
run_cli("migrate", str(src))
config = stored_config(tmp_path)
assert list(config.domains) == ["company.com", "other.com"]
assert config.domains["other.com"].rp_name == "Other"
# Current-format sources are left in place
assert src.is_file()
assert b"migrate:cli:other.com" in (tmp_path / "paskia.kantadb").read_bytes()
def test_migrate_merge_label_combines_rp_ids(run_cli, tmp_path):
"""A multi-domain source merges in one transaction, rp-ids slash-joined."""
run_cli("init", "company.com")
src = write_kantadb(tmp_path, {"one.com": "One", "two.com": "Two"}, name="x.kantadb")
run_cli("migrate", str(src))
assert b"migrate:cli:one.com/two.com" in (tmp_path / "paskia.kantadb").read_bytes()
def test_migrate_merges_shared_domain_origins(run_cli, tmp_path):
"""Same rp-id in both databases: origins union, existing rp-name wins."""
run_cli("init", "example.com", "Existing Name")
src = write_kantadb(tmp_path, {"example.com": "Incoming Name"}, name="x.kantadb")
run_cli("migrate", str(src))
domain = stored_config(tmp_path).domains["example.com"]
assert domain.rp_name == "Existing Name"
assert set(domain.origins) == {"**.example.com"}
def test_migrate_refuses_active_database_as_source(run_cli):
run_cli("init")
with pytest.raises(SystemExit, match="active database"):
run_cli("migrate", "paskia.kantadb")
def test_migrate_kantadb_to_fresh_target(run_cli, tmp_path):
src_dir = tmp_path / "elsewhere"
src_dir.mkdir()
src = write_kantadb(src_dir, {"other.com": "Other"})
run_cli("migrate", str(src))
assert list(stored_config(tmp_path).domains) == ["other.com"]
def test_migrate_without_legacy_database(run_cli):
with pytest.raises(SystemExit, match="No legacy"):