76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
"""Versioned binary BEB codec.
|
|
|
|
The uncompressed header contains the magic bytes and codec version, allowing
|
|
the correct decoder to be selected even if later versions change compression
|
|
or serialization methods.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import zlib
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
|
|
import msgpack
|
|
|
|
FORMAT_NAME = "beb"
|
|
FILE_FORMAT_VERSION = 1
|
|
BEB_MAGIC = b"BEB\x00"
|
|
_VERSION_SIZE = 4
|
|
|
|
|
|
def load_data(path: Path) -> tuple[int, Mapping[str, Any]]:
|
|
"""Read a BEB header and decode its raw document mapping.
|
|
|
|
An unknown version is returned with an empty mapping so the migration
|
|
layer can produce the standard unsupported-version error without trying
|
|
an incompatible decoder.
|
|
"""
|
|
try:
|
|
payload = path.read_bytes()
|
|
if not payload.startswith(BEB_MAGIC):
|
|
raise ValueError("missing BEB file header")
|
|
header_end = len(BEB_MAGIC) + _VERSION_SIZE
|
|
if len(payload) < header_end:
|
|
raise ValueError("truncated BEB file header")
|
|
version = int.from_bytes(payload[len(BEB_MAGIC) : header_end], "big")
|
|
encoded = payload[header_end:]
|
|
decoder = _DECODERS.get(version)
|
|
if decoder is None:
|
|
# The version can be inspected without trying to decompress or
|
|
# deserialize using the wrong algorithm.
|
|
return version, {}
|
|
data = decoder(encoded)
|
|
except (OSError, ValueError, zlib.error, msgpack.exceptions.MsgpackException) as exc:
|
|
raise ValueError(f"could not read BEB document {path}: {exc}") from exc
|
|
if not isinstance(data, Mapping):
|
|
raise ValueError(f"BEB document {path} must contain a map at its root")
|
|
return version, data
|
|
|
|
|
|
def save_data(data: Mapping[str, Any], path: Path) -> None:
|
|
"""Encode raw document data using the current BEB version and write it."""
|
|
encoded = _encode_v1(data)
|
|
version = FILE_FORMAT_VERSION.to_bytes(_VERSION_SIZE, "big")
|
|
path.write_bytes(BEB_MAGIC + version + encoded)
|
|
|
|
|
|
def _encode_v1(data: Mapping[str, Any]) -> bytes:
|
|
"""Encode BEB version 1 as zlib-compressed MessagePack."""
|
|
return zlib.compress(msgpack.packb(dict(data), use_bin_type=True))
|
|
|
|
|
|
def _decode_v1(payload: bytes) -> Any:
|
|
"""Decode a zlib-compressed MessagePack BEB version 1 payload."""
|
|
return msgpack.unpackb(
|
|
zlib.decompress(payload),
|
|
raw=False,
|
|
strict_map_key=False,
|
|
)
|
|
|
|
|
|
# Keep old decoders when adding a new BEB encoding version.
|
|
_DECODERS = {
|
|
1: _decode_v1,
|
|
}
|