refactor: consolidate shared catalog and API helpers

src/govoplan_core/core/configuration_packages.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import base64
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
-from datetime import UTC, datetime
 from pathlib import Path
 import json
 import os
@@ -21,6 +20,7 @@ from govoplan_core.core.module_package_catalog import (
     _is_http_url,
     _load_private_key,
     _parse_trusted_keys,
+    _record_catalog_acceptance,
 )
 from govoplan_core.core.external_references import (
     IntegrationMaturity,
@@ -858,33 +858,10 @@ def sign_configuration_package_catalog(*, path: Path, key_id: str, private_key_p
 
 
 def record_configuration_package_catalog_acceptance(validation: dict[str, object]) -> None:
-    state_path = _configured_sequence_state_path()
+    _record_catalog_acceptance(
-    if state_path is None or validation.get("valid") is not True:
+        validation,
-        return
+        state_path=_configured_sequence_state_path(),
-    channel = validation.get("channel")
+    )
-    sequence = validation.get("sequence")
-    if not isinstance(channel, str) or not isinstance(sequence, int):
-        return
-    try:
-        state = json.loads(state_path.read_text(encoding="utf-8")) if state_path.exists() else {}
-    except json.JSONDecodeError:
-        state = {}
-    if not isinstance(state, dict):
-        state = {}
-    channels = state.get("channels")
-    if not isinstance(channels, dict):
-        channels = {}
-    channel_state = channels.get(channel)
-    if not isinstance(channel_state, dict):
-        channel_state = {}
-    channel_state["last_sequence"] = max(int(channel_state.get("last_sequence") or 0), sequence)
-    channel_state["accepted_at"] = datetime.now(tz=UTC).isoformat().replace("+00:00", "Z")
-    channel_state["key_id"] = validation.get("key_id")
-    channel_state["source"] = validation.get("source") or validation.get("path")
-    channels[channel] = channel_state
-    state["channels"] = channels
-    state_path.parent.mkdir(parents=True, exist_ok=True)
-    state_path.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n", encoding="utf-8")
 
 
 def configuration_package_claim_issues(

src/govoplan_core/core/module_package_catalog.py

@@ -244,7 +244,17 @@ def sign_module_package_catalog(
 
 
 def record_module_package_catalog_acceptance(validation: dict[str, object]) -> None:
-    state_path = _configured_sequence_state_path()
+    _record_catalog_acceptance(
+        validation,
+        state_path=_configured_sequence_state_path(),
+    )
+
+
+def _record_catalog_acceptance(
+    validation: dict[str, object],
+    *,
+    state_path: Path | None,
+) -> None:
     if state_path is None or validation.get("valid") is not True:
         return
     channel = validation.get("channel")

webui/src/api/client.ts

@@ -127,6 +127,19 @@ export function apiPostJson<TResponse, TPayload = unknown>(
   return apiPost<TResponse>(settings, path, { ...init, body: JSON.stringify(payload) });
 }
 
+export function apiPatch<TResponse>(settings: ApiSettings, path: string, init: Omit<RequestInit, "method"> = {}): Promise<TResponse> {
+  return apiFetch<TResponse>(settings, path, { ...init, method: "PATCH" });
+}
+
+export function apiPatchJson<TResponse, TPayload = unknown>(
+  settings: ApiSettings,
+  path: string,
+  payload: TPayload,
+  init: Omit<RequestInit, "method" | "body"> = {}
+): Promise<TResponse> {
+  return apiPatch<TResponse>(settings, path, { ...init, body: JSON.stringify(payload) });
+}
+
 export function loadApiSettings(): ApiSettings {
   const storedBaseUrl = loadStoredSetting("baseUrl");
   const storedApiKey = sessionStorage.getItem(`${SESSION_STORAGE_KEY}.apiKey`);