"""Explicit recursive JSON snapshots for audit/storage, NEVER API write payloads."""
from datetime import date, datetime, time
from enum import Enum
from math import isfinite
from typing import Any
from .data.fields import (
BlobField,
FieldConfigurableObject,
ListField,
ReferenceField,
object_fields,
)
from .data.sobject import SObject
from .json_data import JsonView
[docs]
def serialize_snapshot(
value: Any, *, include_unknown: bool = True, max_depth: int = 100
) -> Any:
"""Keep readonly IDs and optionally captured unknown fields at every depth.
Known fields win name collisions. Does not inspect arbitrary __dict__ values
or serialize clients, credentials, open files or binary data. Snapshots can
still contain sensitive field values; protect them as private audit data.
``serialize_object`` remains the separate write-payload formatter.
"""
if not isinstance(max_depth, int) or isinstance(max_depth, bool):
raise TypeError("Snapshot maximum depth must be an integer")
if max_depth < 0:
raise ValueError("Snapshot maximum depth must not be negative")
active = set()
def visit(current, depth):
if isinstance(current, JsonView):
current = current._data
if depth > max_depth:
raise ValueError("Snapshot exceeds maximum depth")
if isinstance(current, Enum):
return visit(current.value, depth)
if isinstance(current, str):
return str(current)
if current is None or isinstance(current, (bool, int)):
return current
if isinstance(current, float):
if not isfinite(current):
raise ValueError("Snapshots require finite JSON numbers")
return current
if isinstance(current, (datetime, date, time)):
return current.isoformat()
if not isinstance(current, (FieldConfigurableObject, dict, list, tuple)):
raise TypeError("Unsupported snapshot value type")
identity = id(current)
if identity in active:
raise ValueError("Snapshot contains a reference cycle")
active.add(identity)
try:
if isinstance(current, FieldConfigurableObject):
if isinstance(current, SObject):
blob = type(current).attributes.blob_field
if blob is not None and getattr(current, blob, None) is not None:
raise TypeError(
"Binary snapshot fields require explicit handling"
)
result = {}
known = object_fields(type(current))
if include_unknown:
unknown = {
**current.__dict__.get("_unknown_fields", {}),
**current.__dict__.get("additional_fields", {}),
}
if any(not isinstance(name, str) for name in unknown):
raise TypeError("Snapshot mappings require string keys")
for name, raw in unknown.items():
if name in known:
continue
result[name] = visit(raw, depth + 1)
for name, field in known.items():
raw = getattr(current, name, None)
if isinstance(field, BlobField) and raw is not None:
raise TypeError(
"Binary snapshot fields require explicit handling"
)
formatted = (
raw
if raw is None or isinstance(field, (ReferenceField, ListField))
else field.format(raw)
)
result[name] = visit(formatted, depth + 1)
return result
if isinstance(current, dict):
if any(not isinstance(key, str) for key in current):
raise TypeError("Snapshot mappings require string keys")
return {key: visit(item, depth + 1) for key, item in current.items()}
return [visit(item, depth + 1) for item in current]
finally:
active.remove(identity)
return visit(value, 0)