GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

cultcache-py: union codec and nested record codec, held to the C# union-parity golden

Browse source 
define_union and define_record_codec share the slot logic of define_database_entry_type
(one record codec); a field takes a codec, many, or a thunk for recursion. Unknown arm
keys are refused naming union and key, never the payload. All commits build.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-11 03:49:47 +02:00
commit 91c8352ec5
3 changed files with 436 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

8
packages/cultcache-py/src/cultcache_py/__init__.py
View file

@ -4,10 +4,14 @@ from .cache import CultCache, CultCacheBuilder
from .documents import (
DatabaseEntryField,
DocumentDefinition,
SlotCodec,
UnionArm,
database_entry_field,
define_database_entry_type,
define_document_registry,
define_document_type,
define_record_codec,
define_union,
)
from .stores import JsonLinesBackingStore, SingleFileMessagePackBackingStore
@ -22,8 +26,12 @@ __all__ = [
"DocumentDefinition",
"JsonLinesBackingStore",
"SingleFileMessagePackBackingStore",
"SlotCodec",
"UnionArm",
"database_entry_field",
"define_database_entry_type",
"define_document_registry",
"define_document_type",
"define_record_codec",
"define_union",
]

199
packages/cultcache-py/src/cultcache_py/documents.py
View file

@ -2,7 +2,7 @@
import json
from dataclasses import MISSING, dataclass, field
from typing import Any, Callable, Generic, Iterable, TypeVar
from typing import Any, Callable, Generic, Iterable, Mapping, TypeVar
from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember
@ -58,17 +58,46 @@ def _messagepack_payload_decoder(payload: bytes) -> Any:
return msgpack.unpackb(payload, raw=False)
@dataclass(frozen=True)
class SlotCodec:
"""Encodes and decodes the value of one slot: a nested record or a union."""
encode: Callable[[Any], Any]
decode: Callable[[Any], Any]
@dataclass(frozen=True)
class DatabaseEntryField:
name: str
key: int
default: Any = MISSING
codec: SlotCodec | Callable[[], SlotCodec] | None = None
many: bool = False
def database_entry_field(name: str, key: int, *, default: Any = MISSING) -> DatabaseEntryField:
def database_entry_field(
name: str,
key: int,
*,
default: Any = MISSING,
codec: SlotCodec | Callable[[], SlotCodec] | None = None,
many: bool = False,
) -> DatabaseEntryField:
"""A slot. `codec` makes the slot a nested record or union (a zero-argument callable
returning the codec lets a union refer to itself); `many` makes it a list of them."""
if key < 0:
raise ValueError("DatabaseEntry field keys must be non-negative integers")
return DatabaseEntryField(name=name, key=key, default=default)
if many and codec is None:
raise ValueError("A many DatabaseEntry field requires a codec")
return DatabaseEntryField(name=name, key=key, default=default, codec=codec, many=many)
@dataclass(frozen=True)
class UnionArm:
"""A decoded union value: the arm's declared name and its decoded record (None for a unit arm)."""
name: str
value: Any = None
@dataclass(frozen=True)
@ -205,14 +234,7 @@ def define_database_entry_type(
and newly added fields can use higher keys without rewriting older payloads.
"""
normalized = [_normalize_database_entry_field(field) for field in fields]
if not normalized:
raise ValueError("DatabaseEntry document types require at least one field")
keys = [field.key for field in normalized]
if len(keys) != len(set(keys)):
raise ValueError(f"DatabaseEntry document type {type!r} has duplicate field keys")
normalized.sort(key=lambda field: field.key)
max_key = max(keys)
normalized = _normalize_fields(type, fields)
member_name_lookup = name if isinstance(name, str) else None
index_lookup = indexes or {}
members = tuple(
@ -229,36 +251,12 @@ def define_database_entry_type(
for field in normalized
)
def encode(value: Any) -> list[Any]:
slots = [None] * (max_key + 1)
for field in normalized:
slots[field.key] = extract_value(value, field.name)
return slots
def decode(raw: Any) -> Any:
if not isinstance(raw, list):
raise ValueError(f"DatabaseEntry document type {type!r} expected a MessagePack array")
values: dict[str, Any] = {}
for field in normalized:
if field.key < len(raw):
value = raw[field.key]
elif field.default is not MISSING:
value = field.default() if callable(field.default) else field.default
else:
raise ValueError(
f"DatabaseEntry document type {type!r} is missing required slot {field.key}"
)
if value is None and field.default is not MISSING:
value = field.default() if callable(field.default) else field.default
values[field.name] = value
if cls is None:
return values
return cls(**values)
record = _record_codec(type, normalized, cls)
return define_document_type(
type,
encode=encode,
decode=decode,
encode=record.encode,
decode=record.decode,
validate=validate,
global_document=global_document,
name=name,
@ -275,6 +273,60 @@ def define_database_entry_type(
)
def define_record_codec(
name: str,
fields: Iterable[DatabaseEntryField | tuple[str, int] | tuple[str, int, Any]],
*,
cls: type[Any] | None = None,
) -> SlotCodec:
"""A nested record: the same slot array as a document payload, usable as a field's codec or a union arm."""
return _record_codec(name, _normalize_fields(name, fields), cls)
def define_union(
name: str,
arms: Mapping[int, tuple[str, SlotCodec | Callable[[], SlotCodec] | None]],
) -> SlotCodec:
"""A union: nil, or `[armKey, [armSlots...]]`. Each arm is `(arm name, record codec)`; a unit arm has
no codec and is written `[armKey, []]`. A key the union does not declare is refused, never carried."""
by_key = dict(arms)
by_name = {arm_name: (key, codec) for key, (arm_name, codec) in by_key.items()}
if not by_key:
raise ValueError(f"Union {name!r} requires at least one arm")
if len(by_name) != len(by_key):
raise ValueError(f"Union {name!r} has duplicate arm names")
if any(not _is_key(key) or key < 0 for key in by_key):
raise ValueError(f"Union {name!r} arm keys must be non-negative integers")
def encode(value: Any) -> Any:
if value is None:
return None
if not isinstance(value, UnionArm) or value.name not in by_name:
raise ValueError(f"Union {name!r} was given a value that is not one of its arms")
key, codec = by_name[value.name]
if codec is None:
if value.value is not None:
raise ValueError(f"Union {name!r} arm {key} is a unit arm and carries no value")
return [key, []]
return [key, _resolve(codec).encode(value.value)]
def decode(raw: Any) -> Any:
if raw is None:
return None
if not isinstance(raw, list) or len(raw) != 2 or not isinstance(raw[1], list):
raise ValueError(f"Union {name!r} value is malformed: expected [armKey, [armSlots]]")
key = raw[0]
if not _is_key(key):
raise ValueError(f"Union {name!r} value is malformed: the arm key is not an integer")
if key not in by_key:
declared = ", ".join(str(declared_key) for declared_key in sorted(by_key))
raise ValueError(f"Union {name!r} does not declare arm key {key} (declared: {declared})")
arm_name, codec = by_key[key]
return UnionArm(arm_name, None if codec is None else _resolve(codec).decode(raw[1]))
return SlotCodec(encode=encode, decode=decode)
def define_document_registry(*documents: DocumentDefinition[Any]) -> tuple[DocumentDefinition[Any], ...]:
return tuple(documents)
@ -287,6 +339,77 @@ def extract_value(value: Any, extractor: str | Extractor[Any]) -> str | int | fl
return getattr(value, extractor, None)
def _is_key(value: Any) -> bool:
return isinstance(value, int) and not isinstance(value, bool)
def _resolve(codec: SlotCodec | Callable[[], SlotCodec]) -> SlotCodec:
return codec if isinstance(codec, SlotCodec) else codec()
def _normalize_fields(
type: str,
fields: Iterable[DatabaseEntryField | tuple[str, int] | tuple[str, int, Any]],
) -> list[DatabaseEntryField]:
normalized = [_normalize_database_entry_field(field) for field in fields]
if not normalized:
raise ValueError("DatabaseEntry document types require at least one field")
keys = [field.key for field in normalized]
if len(keys) != len(set(keys)):
raise ValueError(f"DatabaseEntry document type {type!r} has duplicate field keys")
return sorted(normalized, key=lambda field: field.key)
def _record_codec(type: str, normalized: list[DatabaseEntryField], cls: type[Any] | None) -> SlotCodec:
max_key = normalized[-1].key
def encode_slot(field: DatabaseEntryField, value: Any) -> Any:
if field.codec is None or value is None:
return value
codec = _resolve(field.codec)
if field.many:
return [None if item is None else codec.encode(item) for item in value]
return codec.encode(value)
def decode_slot(field: DatabaseEntryField, value: Any) -> Any:
if field.codec is None or value is None:
return value
codec = _resolve(field.codec)
if field.many:
if not isinstance(value, list):
raise ValueError(f"DatabaseEntry document type {type!r} slot {field.key} expected a MessagePack array")
return [None if item is None else codec.decode(item) for item in value]
return codec.decode(value)
def encode(value: Any) -> list[Any]:
slots = [None] * (max_key + 1)
for field in normalized:
slots[field.key] = encode_slot(field, extract_value(value, field.name))
return slots
def decode(raw: Any) -> Any:
if not isinstance(raw, list):
raise ValueError(f"DatabaseEntry document type {type!r} expected a MessagePack array")
values: dict[str, Any] = {}
for field in normalized:
if field.key < len(raw):
value = decode_slot(field, raw[field.key])
elif field.default is not MISSING:
value = field.default() if callable(field.default) else field.default
else:
raise ValueError(
f"DatabaseEntry document type {type!r} is missing required slot {field.key}"
)
if value is None and field.default is not MISSING:
value = field.default() if callable(field.default) else field.default
values[field.name] = value
if cls is None:
return values
return cls(**values)
return SlotCodec(encode=encode, decode=decode)
def _normalize_database_entry_field(
field: DatabaseEntryField | tuple[str, int] | tuple[str, int, Any],
) -> DatabaseEntryField:

267
packages/cultcache-py/tests/test_union_parity.py Normal file
View file

@ -0,0 +1,267 @@
from __future__ import annotations
import shutil
import tempfile
import unittest
from pathlib import Path
import msgpack
from cultcache_py import (
CultCache,
SingleFileMessagePackBackingStore,
UnionArm,
database_entry_field,
define_database_entry_type,
define_record_codec,
define_union,
)
VECTORS = Path(__file__).resolve().parents[3] / "tests" / "vectors" / "union-parity"
composition = define_record_codec(
"Composition", [("name", 0), database_entry_field("body", 1, codec=lambda: node)]
)
box = define_record_codec("Box", [("id", 0), database_entry_field("children", 1, codec=lambda: node, many=True)])
text = define_record_codec("Text", [("id", 0), ("content", 1)])
use = define_record_codec(
"Use",
[
("id", 0),
database_entry_field("composition", 1, codec=composition),
database_entry_field("slots", 2, codec=lambda: node, many=True),
],
)
image = define_record_codec("Image", [("id", 0), ("uri", 1), ("alt", 3)])
node = define_union("Node", {0: ("Box", box), 1: ("Text", text), 2: ("Use", use), 5: ("Image", image)})
outcome = define_union(
"Outcome",
{
0: ("Pending", define_record_codec("Pending", [("message", 0)])),
1: ("Completed", define_record_codec("Completed", [("source_version", 0)])),
2: ("Denied", None),
},
)
tool = define_record_codec("Tool", [("name", 0), ("uses", 1)])
part = define_record_codec("Part", [("name", 0), ("count", 1)])
gem = define_record_codec("Gem", [("name", 0)])
item = define_union("Item", {0: ("Gem", gem), 1: ("Tool", tool), 2: ("Part", part)})
crafted = define_union("CraftedItem", {0: ("Tool", tool), 1: ("Part", part)})
surface_document = define_database_entry_type(
"tests.union_surface",
[
("id", 0),
("version", 1),
database_entry_field("root", 2, codec=node),
database_entry_field("outcomes", 3, codec=outcome, many=True),
database_entry_field("part", 4, codec=item),
database_entry_field("crafted", 5, codec=crafted),
database_entry_field("missing", 6, codec=item),
],
)
def _surface_value() -> dict:
return {
"id": "surface",
"version": 3,
"root": UnionArm(
"Box",
{
"id": "root",
"children": [
UnionArm("Text", {"id": "t1", "content": "hello"}),
UnionArm(
"Use",
{
"id": "u1",
"composition": {
"name": "comp",
"body": UnionArm(
"Box",
{
"id": "inner",
"children": [
UnionArm("Text", {"id": "deep", "content": "x"}),
UnionArm("Image", {"id": "img", "uri": "uri://a", "alt": "alt"}),
],
},
),
},
"slots": [
UnionArm("Text", {"id": "s1", "content": "slot"}),
None,
UnionArm("Box", {"id": "b2", "children": []}),
],
},
),
UnionArm("Image", {"id": "i2", "uri": "uri://b", "alt": ""}),
],
},
),
"outcomes": [
UnionArm("Pending", {"message": "wait"}),
UnionArm("Completed", {"source_version": 7}),
UnionArm("Denied"),
],
"part": UnionArm("Tool", {"name": "hammer", "uses": 12}),
"crafted": UnionArm("Part", {"name": "gear", "count": 4}),
"missing": None,
}
def _vector(name: str) -> bytes:
return (VECTORS / name).read_bytes()
class UnionSurfaceVectorTests(unittest.TestCase):
def test_surface_payload_decodes_to_the_value_and_re_encodes_byte_identical(self) -> None:
payload = _vector("union-surface.msgpack")
decoded = surface_document.decode_payload(payload)
self.assertEqual(decoded, _surface_value())
self.assertEqual(surface_document.encode_payload(decoded), payload)
def test_the_csharp_written_store_reads_and_the_cache_rewrites_the_same_payload(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "union-surface.cc"
shutil.copyfile(VECTORS / "union-surface.cc", path)
cache = (
CultCache.builder()
.register_document_type(surface_document)
.add_generic_store(SingleFileMessagePackBackingStore(path))
.build()
)
cache.pull_all_backing_stores()
self.assertEqual(cache.get(surface_document, "surface"), _surface_value())
fresh = Path(tmp) / "fresh.cc"
writer = (
CultCache.builder()
.register_document_type(surface_document)
.add_generic_store(SingleFileMessagePackBackingStore(fresh))
.build()
)
writer.pull_all_backing_stores()
writer.put(surface_document, "surface", _surface_value())
[envelope] = SingleFileMessagePackBackingStore(fresh).pull_all()
self.assertEqual(envelope.payload, _vector("union-surface.msgpack"))
def test_the_store_vector_payload_equals_the_surface_payload(self) -> None:
[envelope] = SingleFileMessagePackBackingStore(VECTORS / "union-surface.cc").pull_all()
self.assertEqual(envelope.payload, _vector("union-surface.msgpack"))
def test_every_outcome_in_the_manifest_is_held(self) -> None:
outcomes = msgpack.unpackb(_vector("outcomes.msgpack"), raw=False)
self.assertEqual(len(outcomes), 9)
refused = read_only = 0
for file, outcome_kind, union, key, detail in outcomes:
payload = _vector(file)
if outcome_kind == "refused":
refused += 1
with self.assertRaises(ValueError, msg=file) as caught:
surface_document.decode_payload(payload)
message = str(caught.exception)
self.assertIn(union, message, file)
if detail == "unknown-arm":
self.assertIn(f"arm key {key} ", message, file)
else:
self.assertEqual(detail, "arity")
self.assertIn("malformed", message, file)
else:
read_only += 1
self.assertEqual(outcome_kind, "read-only", file)
self.assertEqual(surface_document.decode_payload(payload), _surface_value(), file)
self.assertEqual((refused, read_only), (5, 4))
class UnionCodecRuleTests(unittest.TestCase):
CANARY = "canary-7f3a91"
def test_nil_union_is_nil_both_ways(self) -> None:
self.assertIsNone(node.encode(None))
self.assertIsNone(node.decode(None))
def test_unit_arm_is_written_with_empty_slots_and_reads_back(self) -> None:
self.assertEqual(outcome.encode(UnionArm("Denied")), [2, []])
self.assertEqual(outcome.decode([2, []]), UnionArm("Denied", None))
def test_unit_arm_refuses_a_value_to_carry(self) -> None:
with self.assertRaises(ValueError) as caught:
outcome.encode(UnionArm("Denied", {"message": self.CANARY}))
self.assertIn("Outcome", str(caught.exception))
self.assertNotIn(self.CANARY, str(caught.exception))
def test_arm_keys_are_written_as_declared(self) -> None:
self.assertEqual(node.encode(UnionArm("Image", {"id": "a", "uri": "b", "alt": "c"})), [5, ["a", "b", None, "c"]])
self.assertEqual(node.encode(UnionArm("Text", {"id": "a", "content": "b"})), [1, ["a", "b"]])
def test_encode_refuses_an_arm_the_union_does_not_declare_without_echoing_it(self) -> None:
with self.assertRaises(ValueError) as caught:
crafted.encode(UnionArm("Gem", {"name": self.CANARY}))
self.assertIn("CraftedItem", str(caught.exception))
self.assertNotIn(self.CANARY, str(caught.exception))
with self.assertRaises(ValueError):
node.encode({"id": "a"})
def test_unknown_key_names_the_union_the_key_and_the_declared_keys_never_the_payload(self) -> None:
with self.assertRaises(ValueError) as caught:
node.decode([4, [self.CANARY]])
message = str(caught.exception)
self.assertIn("'Node'", message)
self.assertIn("arm key 4 ", message)
self.assertIn("declared: 0, 1, 2, 5", message)
self.assertNotIn(self.CANARY, message)
def test_a_malformed_union_value_is_refused_without_echoing_it(self) -> None:
for raw in (
[1],
[1, ["a"], "extra"],
[1, self.CANARY],
self.CANARY,
{"k": self.CANARY},
[self.CANARY, []],
[True, []],
[1.0, []],
):
with self.subTest(raw=type(raw).__name__):
with self.assertRaises(ValueError) as caught:
node.decode(raw)
self.assertIn("'Node'", str(caught.exception))
self.assertNotIn(self.CANARY, str(caught.exception))
def test_extra_trailing_arm_slots_are_read(self) -> None:
self.assertEqual(
node.decode([1, ["a", "b", "extra"]]),
UnionArm("Text", {"id": "a", "content": "b"}),
)
def test_many_slot_requires_a_list_and_keeps_nil_elements(self) -> None:
slots = [UnionArm("Text", {"id": "a", "content": "b"}), None]
encoded = use.encode({"id": "u", "composition": None, "slots": slots})
self.assertEqual(encoded, ["u", None, [[1, ["a", "b"]], None]])
self.assertEqual(use.decode(encoded)["slots"], slots)
with self.assertRaises(ValueError) as caught:
use.decode(["u", None, self.CANARY])
self.assertNotIn(self.CANARY, str(caught.exception))
def test_a_nested_record_with_a_missing_required_slot_is_refused(self) -> None:
with self.assertRaises(ValueError) as caught:
node.decode([1, ["only-id"]])
self.assertIn("Text", str(caught.exception))
self.assertIn("slot 1", str(caught.exception))
def test_definition_rules(self) -> None:
with self.assertRaises(ValueError):
define_union("Empty", {})
with self.assertRaises(ValueError):
define_union("Dup", {0: ("A", None), 1: ("A", None)})
with self.assertRaises(ValueError):
define_union("Neg", {-1: ("A", None)})
with self.assertRaises(ValueError):
define_union("Flag", {True: ("A", None), "x": ("B", None)})
with self.assertRaises(ValueError):
database_entry_field("loose", 1, many=True)
if __name__ == "__main__":
unittest.main()