31 lines
1.4 KiB
Python
31 lines
1.4 KiB
Python
"""Read-only operator migration records for pre-ledger task branches."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Callable
|
|
|
|
from gitea_api_policy import PolicyError
|
|
from scm_task_grants import ADOPTIONS_PATH, TaskLedger
|
|
|
|
|
|
def seed(ledger: TaskLedger, token: str, branch_head: Callable[[str, str, str], str | None]) -> None:
|
|
"""Seed each independently reviewed record without blocking unrelated SCM."""
|
|
try:
|
|
records = json.loads(ADOPTIONS_PATH.read_bytes())
|
|
except (OSError, TypeError, ValueError):
|
|
print("SCM task adoption registry is unavailable; deferring migration", flush=True)
|
|
return
|
|
if not isinstance(records, dict) or len(records) > 64:
|
|
print("SCM task adoption registry is invalid; deferring migration", flush=True)
|
|
return
|
|
for key, record in records.items():
|
|
try:
|
|
if not isinstance(key, str) or not isinstance(record, dict):
|
|
raise PolicyError("task branch adoption registry is invalid")
|
|
repo, ref = record.get("repo"), record.get("ref")
|
|
if key != f"{repo}/{ref}":
|
|
raise PolicyError("task branch adoption registry is invalid")
|
|
ledger.seed_adoption(record, branch_head(repo, ref, token))
|
|
except (OSError, PolicyError, TypeError, ValueError):
|
|
print(f"SCM task adoption deferred for {key!r}", flush=True)
|