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

Resolve Python snapshot discovery by schema alias

Browse source
This commit is contained in:
Meta Vens 2026-06-27 21:42:32 +01:00
commit bff2e552c8
3 changed files with 109 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

30
packages/cultcache-py/src/cultmesh_py/node.py
View file

@ -535,6 +535,8 @@ class CultMeshDatabase:
) -> CultNetRawSnapshotResponse:
requested_schema_ids = set(schema_ids or [])
requested_record_keys = set(record_keys or [])
documents_by_type = {document.type: document for document in self.documents}
documents_by_schema_id = schema_document_map(self.documents)
shard_record_keys = (
self._live_shard_record_keys(shard_id)
if shard_id is not None and shard_id in self._shard_logs
@ -543,7 +545,8 @@ class CultMeshDatabase:
documents: list[CultNetRawDocumentRecord] = []
for envelope in self.cache.snapshot_envelopes():
schema_id = envelope.schema_id or envelope.type
if requested_schema_ids and schema_id not in requested_schema_ids:
document = documents_by_type.get(envelope.type) or documents_by_schema_id.get(schema_id)
if requested_schema_ids and not self._schema_matches_request(schema_id, document, requested_schema_ids):
continue
if requested_record_keys and envelope.key not in requested_record_keys:
continue
@ -950,6 +953,31 @@ class CultMeshDatabase:
previous[(resolved_schema_id, record_key)] = self.cache.get(document, record_key)
return previous
@staticmethod
def _schema_matches_request(
schema_id: str,
document: DocumentDefinition[Any] | None,
requested_schema_ids: set[str],
) -> bool:
if not requested_schema_ids or schema_id in requested_schema_ids:
return True
if document is None:
return False
entry = document.catalog_entry()
if entry.schema_id in requested_schema_ids or entry.schema_name in requested_schema_ids:
return True
if any(schema_id in requested_schema_ids for schema_id in entry.compatible_schema_ids):
return True
return any(_infer_schema_name(schema_id) == entry.schema_name for schema_id in requested_schema_ids)
def _infer_schema_name(schema_id: str) -> str | None:
marker = schema_id.rfind(".v")
if marker <= 0 or marker + 2 >= len(schema_id):
return None
version = schema_id[marker + 2:]
return schema_id[:marker] if version.isdigit() else None
def create_node(
cache_path: str | Path | None = None,

47
packages/cultcache-py/src/cultmesh_py/server.py
View file

@ -591,12 +591,21 @@ class CultMeshLocalServer:
def _shard_catalog_response(self, request: dict[str, Any]) -> dict[str, Any]:
requested_schema_ids = {str(value) for value in request.get("schemaIds") or []}
document_schema_ids = [document.catalog_entry().schema_id for document in self.node.documents]
documents_by_schema_id = _schema_document_map(self.node.documents)
shard_ids = self.node.database.shard_ids() or ["primary"]
shards = []
for shard_id in shard_ids:
schema_ids = self.node.database.shard_schema_ids(shard_id) or document_schema_ids
if requested_schema_ids:
schema_ids = [schema_id for schema_id in schema_ids if schema_id in requested_schema_ids]
schema_ids = [
schema_id
for schema_id in schema_ids
if _schema_matches_request(
schema_id,
documents_by_schema_id.get(schema_id),
requested_schema_ids,
)
]
if not schema_ids:
continue
shards.append({
@ -620,3 +629,39 @@ class CultMeshLocalServer:
def _now_ms() -> int:
return int(time.time() * 1000)
def _schema_document_map(documents: list[Any]) -> dict[str, Any]:
mapped: dict[str, Any] = {}
for document in documents:
entry = document.catalog_entry()
mapped[entry.schema_id] = document
mapped[entry.schema_name] = document
for compatible_schema_id in entry.compatible_schema_ids:
mapped[compatible_schema_id] = document
return mapped
def _schema_matches_request(
schema_id: str,
document: Any | None,
requested_schema_ids: set[str],
) -> bool:
if not requested_schema_ids or schema_id in requested_schema_ids:
return True
if document is None:
return False
entry = document.catalog_entry()
if entry.schema_id in requested_schema_ids or entry.schema_name in requested_schema_ids:
return True
if any(compatible_schema_id in requested_schema_ids for compatible_schema_id in entry.compatible_schema_ids):
return True
return any(_infer_schema_name(requested_schema_id) == entry.schema_name for requested_schema_id in requested_schema_ids)
def _infer_schema_name(schema_id: str) -> str | None:
marker = schema_id.rfind(".v")
if marker <= 0 or marker + 2 >= len(schema_id):
return None
version = schema_id[marker + 2:]
return schema_id[:marker] if version.isdigit() else None

34
packages/cultcache-py/tests/test_cultcache.py
View file

@ -5233,6 +5233,40 @@ class CultCacheTests(unittest.TestCase):
self.assertIsNone(subscription_delete.raw_document)
self.assertIsNone(node.database.get(document, "note:2"))
def test_cultmesh_local_server_resolves_snapshot_and_shard_catalog_schema_aliases(self) -> None:
document = define_database_entry_type(
"mesh.alias_server_note",
[("body", 0)],
schema_id="sha256:mesh-alias-server-note",
schema_name="mesh.alias_server_note",
schema_version="mesh.alias_server_note.v1",
)
node = CultMesh.create_node(runtime_id="mesh-alias-server")
node.database.register_document(document)
node.database.put_raw_message(
document,
"note:alias",
{"body": "served through alias"},
shard_id="notes",
shard_epoch=1,
)
server = CultMesh.serve_node(node)
try:
raw_client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0)
snapshot = raw_client.fetch_snapshot(schema_ids=["mesh.alias_server_note.v1"])
typed_snapshot = raw_client.fetch_snapshot_response(schema_ids=["mesh.alias_server_note.v1"])
shard_catalog = raw_client.fetch_shard_catalog(schema_ids=["mesh.alias_server_note.v1"])
finally:
server.stop()
local_schema_id = document.catalog_entry().schema_id
self.assertEqual(snapshot["documents"][0]["schemaId"], local_schema_id)
self.assertEqual(snapshot["documents"][0]["recordKey"], "note:alias")
self.assertEqual(typed_snapshot.documents[0].schema_id, local_schema_id)
self.assertEqual(shard_catalog["shards"][0]["schemaIds"], [local_schema_id])
self.assertEqual(shard_catalog["shards"][0]["shardId"], "notes")
def test_cultmesh_local_server_rejects_oversized_snapshot_responses(self) -> None:
document = define_database_entry_type(
"mesh.snapshot_limit_note",