29 lines
1.2 KiB
Python
29 lines
1.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
from __future__ import annotations
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
VALID_SESSION_STATUSES = {"planned", "in_progress", "stopped", "completed"}
|
|
VALID_ITEM_STATUSES = {"planned", "completed", "partial", "skipped"}
|
|
|
|
|
|
def merge_session_patch(existing: dict[str, Any], payload: dict[str, Any], now: str) -> dict[str, Any]:
|
|
current_revision = int(existing.get("revision") or 1)
|
|
expected = payload.get("expected_revision")
|
|
if expected is not None and int(expected) != current_revision:
|
|
raise RuntimeError(f"revision_conflict:{current_revision}")
|
|
result = deepcopy(existing)
|
|
profile = payload.get("profile")
|
|
if isinstance(profile, dict):
|
|
result["profile"] = deepcopy(profile)
|
|
week_statuses = payload.get("week_statuses")
|
|
if isinstance(week_statuses, dict):
|
|
result["week_statuses"] = deepcopy(week_statuses)
|
|
key = str(payload.get("session_key") or "")
|
|
session = payload.get("session")
|
|
if key and isinstance(session, dict):
|
|
result.setdefault("sessions", {})[key] = deepcopy(session)
|
|
result["revision"] = current_revision + 1
|
|
result["updated_at"] = now
|
|
return result
|