Cache Python document schema catalogs

This commit is contained in:
Meta Vens 2026-06-14 01:43:14 +01:00
commit 87f7868af1
2 changed files with 21 additions and 1 deletions

View file

@ -89,6 +89,7 @@ class DocumentDefinition(Generic[T]):
canonical_schema_json: str | None = None
compatible_schema_ids: tuple[str, ...] = field(default_factory=tuple)
members: tuple[CultCacheSchemaCatalogMember, ...] = field(default_factory=tuple)
_catalog_entry: CultCacheSchemaCatalogEntry | None = field(default=None, init=False, repr=False, compare=False)
def encode_payload(self, value: T) -> bytes:
if self.validate:
@ -102,6 +103,8 @@ class DocumentDefinition(Generic[T]):
return value
def catalog_entry(self) -> CultCacheSchemaCatalogEntry:
if self._catalog_entry is not None:
return self._catalog_entry
schema_id = self.schema_id or self.type
schema_name = self.schema_name or self.type
schema_version = self.schema_version or f"{schema_name}.v1"
@ -127,7 +130,7 @@ class DocumentDefinition(Generic[T]):
separators=(",", ":"),
)
compatible_schema_ids = self.compatible_schema_ids or (schema_id,)
return CultCacheSchemaCatalogEntry(
catalog_entry = CultCacheSchemaCatalogEntry(
schema_id=schema_id,
schema_name=schema_name,
schema_version=schema_version,
@ -136,6 +139,8 @@ class DocumentDefinition(Generic[T]):
compatible_schema_ids=tuple(compatible_schema_ids),
members=tuple(self.members),
)
object.__setattr__(self, "_catalog_entry", catalog_entry)
return catalog_entry
def define_document_type(

View file

@ -190,6 +190,21 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(msgpack.unpackb(payload, raw=False), ["ash", None, 3])
self.assertEqual(document.decode_payload(msgpack.packb(["ash"], use_bin_type=True)).retries, 0)
def test_document_catalog_entry_is_cached_after_first_derivation(self) -> None:
document = define_database_entry_type(
"settings",
[
("theme", 0),
("retries", 2, 0),
],
)
first = document.catalog_entry()
second = document.catalog_entry()
self.assertIs(first, second)
self.assertEqual([member.slot for member in first.members], [0, 2])
def test_messagepack_store_writes_v1_snapshot(self) -> None:
try:
import msgpack # type: ignore