Migration script: Fix error fetching /api/principal in community edition (x2)

This commit is contained in:
Maurus Decimus
2026-04-20 20:08:37 +02:00
parent e79f1b070d
commit ab0cfd5bad

View File

@@ -168,7 +168,23 @@ class StalwartClient:
def list_principal_names(self) -> list[tuple[str, str]]: def list_principal_names(self) -> list[tuple[str, str]]:
out: list[tuple[str, str]] = [] out: list[tuple[str, str]] = []
seen: set[tuple[str, str]] = set() seen: set[tuple[str, str]] = set()
types_param = ",".join(ALL_PRINCIPAL_TYPES) for principal_type in ALL_PRINCIPAL_TYPES:
try:
self._list_principals_of_type(principal_type, out, seen)
except ApiError as exc:
print(
f" WARN skipping principal type {principal_type!r}: {exc}",
file=sys.stderr,
)
return out
def _list_principals_of_type(
self,
principal_type: str,
out: list[tuple[str, str]],
seen: set[tuple[str, str]],
) -> None:
before = len(out)
page = 1 page = 1
while True: while True:
data = self.get( data = self.get(
@@ -176,13 +192,13 @@ class StalwartClient:
params={ params={
"page": str(page), "page": str(page),
"limit": str(PAGE_SIZE), "limit": str(PAGE_SIZE),
"types": types_param, "types": principal_type,
}, },
) )
items = data.get("items", []) or [] items = data.get("items", []) or []
total = int(data.get("total", 0) or 0) total = int(data.get("total", 0) or 0)
for p in items: for p in items:
typ = p.get("type") or "" typ = p.get("type") or principal_type
name = _principal_name(p) name = _principal_name(p)
if not name: if not name:
continue continue
@@ -193,10 +209,9 @@ class StalwartClient:
out.append(key) out.append(key)
if not items: if not items:
break break
if len(out) >= total: if (len(out) - before) >= total:
break break
page += 1 page += 1
return out
def get_principal(self, name: str) -> dict[str, Any]: def get_principal(self, name: str) -> dict[str, Any]:
quoted = urllib.parse.quote(name, safe="") quoted = urllib.parse.quote(name, safe="")
@@ -239,18 +254,10 @@ def cmd_dump(args: argparse.Namespace) -> int:
file=sys.stderr) file=sys.stderr)
print("Listing principals...", file=sys.stderr) print("Listing principals...", file=sys.stderr)
names: list[tuple[str, str]] = [] names = client.list_principal_names()
principals: list[dict[str, Any]] = [] print(f" found {len(names)} principals across all types", file=sys.stderr)
try:
names = client.list_principal_names()
except ApiError as exc:
print(f" WARN could not list principals: {exc}", file=sys.stderr)
print(" skipping principal dump (continuing without principals)",
file=sys.stderr)
else:
print(f" found {len(names)} principals across all types",
file=sys.stderr)
principals: list[dict[str, Any]] = []
for i, (typ, name) in enumerate(names, 1): for i, (typ, name) in enumerate(names, 1):
try: try:
full = client.get_principal(name) full = client.get_principal(name)