diff --git a/.gitignore b/.gitignore index 4a48d6a9..7c782540 100644 --- a/.gitignore +++ b/.gitignore @@ -301,6 +301,8 @@ node_modules/ !packages/cultnet-ts/** !packages/cultmesh-ts/ !packages/cultmesh-ts/** +!packages/cultcache-py/ +!packages/cultcache-py/** !packages/cultcache-rs/ !packages/cultcache-rs/** !packages/cultnet-rs/ diff --git a/README.md b/README.md index fa0d8cd2..64109922 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,7 @@ The solution includes: - `packages/cultcache-ts`: TypeScript CultCache with MessagePack persistence and inspector tooling - `packages/cultnet-ts`: TypeScript CultNet schema-v0 contracts, framing, discovery, raw document replication, and interop tests - `packages/cultmesh-ts`: TypeScript CultMesh local node and mesh catalog surface for local runtimes such as VoidBot +- `packages/cultcache-py`: Python CultCache/CultNet/CultMesh package with CultCache v1 wire parity - `packages/cultcache-rs`: Rust CultCache and derive macro - `packages/cultnet-rs`: Rust CultNet contracts, framing, discovery, and interop peer @@ -93,6 +94,7 @@ packages/ cultcache-ts/ cultnet-ts/ cultmesh-ts/ + cultcache-py/ cultcache-rs/ cultnet-rs/ ``` diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md new file mode 100644 index 00000000..e4e5ce1f --- /dev/null +++ b/packages/cultcache-py/README.md @@ -0,0 +1,175 @@ +# cultcache-py + +`cultcache-py` is the Python runtime package for CultCache, CultNet, and +CultMesh. Callers work with registered domain documents while the cache owns +schema identity, routing, globals, name lookups, indexes, and backing-store +envelopes. + +It is intentionally small. It is not an ORM, a database, or distributed consensus wearing rented authority. + +## Current Shape + +- document types are registered explicitly +- `SingleFileMessagePackBackingStore` writes `cultcache.store.v1` snapshots: + `[format, schemaCatalog, records]` +- records store raw MessagePack payload bytes under schema-catalog identity +- payloads decode only through the registered document definition for that type +- unknown persisted types fail closed +- global documents are singleton-style per type +- type-specific backing stores beat generic backing stores +- the JSONL store uses only the Python standard library for bootstrap consumers +- `define_database_entry_type(...)` emits Rust/C#-style slot-indexed + MessagePack array payloads for cross-runtime `DatabaseEntry` contracts +- `cultnet_py` exposes schema-v0 MessagePack helpers and 4-byte big-endian + frame helpers +- `cultmesh_py` exposes a cache-backed local node surface for Python tools + +## Example + +```python +from dataclasses import dataclass, asdict +from cultcache_py import CultCache, JsonLinesBackingStore, define_document_type + +@dataclass +class Settings: + theme: str + retries: int + +settings_doc = define_document_type( + "settings", + encode=lambda value: asdict(value), + decode=lambda payload: Settings(**payload), + global_document=True, +) + +cache = ( + CultCache.builder() + .register_document_type(settings_doc) + .add_generic_store(JsonLinesBackingStore("state.cultcache.jsonl")) + .build() +) + +cache.pull_all_backing_stores() +cache.put_global(settings_doc, Settings(theme="ash", retries=3)) +settings = cache.get_required_global(settings_doc) +``` + +## Public Surface + +- `define_document_type(...)` +- `define_database_entry_type(...)` +- `database_entry_field(...)` +- `define_document_registry(...)` +- `CultCache.builder()` +- `register_document_type(...)` +- `register_registry(...)` +- `register_name_lookup(...)` +- `register_index(...)` +- `add_backing_store(...)` +- `add_generic_store(...)` +- `pull_all_backing_stores()` +- `get(...)` +- `get_required(...)` +- `get_all(...)` +- `get_key_by_name(...)` +- `get_by_name(...)` +- `get_key_by_index(...)` +- `get_by_index(...)` +- `get_global(...)` +- `get_required_global(...)` +- `put(...)` +- `put_global(...)` +- `update(...)` +- `update_global(...)` +- `delete(...)` +- `delete_global(...)` +- `snapshot()` + +## Persistence Model + +Backing stores persist envelopes, not caller domain objects. The cache decodes payloads through registered document definitions and rejects unknown type discriminators. This keeps Python's dynamic runtime from turning persistence into an open polymorphic sewer with a cheerful docstring. + +`JsonLinesBackingStore` is the dependency-free control-plane store. It rewrites an atomic JSONL snapshot and base64-encodes payload bytes. It is designed for compact state spines, settings, ledgers, and bootstrap surfaces. + +`SingleFileMessagePackBackingStore` follows the shared CultCache v1 store shape +used by the TypeScript, Rust, and C# runtimes: + +- top-level MessagePack value: `[format, schemaCatalog, records]` +- `format`: `cultcache.store.v1` +- catalog entries: `[schemaId, schemaName, schemaVersion, contentHash, + canonicalSchemaJson, compatibleSchemaIds, members]` +- records: `[key, schemaId, storedAt, payload]` + +Large corpora should use a sharded store or a real database. A single snapshot file is a scalpel, not a forklift. + +## DatabaseEntry Slot Contract + +For cross-runtime cache entries, use `define_database_entry_type(...)` instead +of the generic JSON formatter: + +```python +from dataclasses import dataclass +from cultcache_py import define_database_entry_type + +@dataclass +class Settings: + theme: str + retries: int = 0 + +settings_doc = define_database_entry_type( + "settings", + [ + ("theme", 0), + ("retries", 1, 0), + ], + cls=Settings, +) +``` + +The payload is a MessagePack array. Field keys are durable slot indexes: + +- key `0` writes array slot 0 +- key `4` writes array slot 4 +- unused slots are written as nil +- deleted fields should leave their slots reserved +- newly added fields should use new keys +- fields with defaults tolerate missing or nil slots when older payloads are read + +That matches the Rust `#[derive(DatabaseEntry)]` formatter shape and the C# +`[Key(n)]` intent: schema evolution comes from stable field slots, not source +member order. + +## CultNet And CultMesh + +`cultnet_py` provides the Python schema-v0 wire helpers: + +```python +from cultnet_py import decode_frame, encode_frame, hello, parse_message + +payload = hello(runtime_id="python-runtime").to_bytes() +message = parse_message(decode_frame(encode_frame(payload))) +``` + +`cultmesh_py` starts with a local cache-backed node: + +```python +from cultcache_py import define_database_entry_type +from cultmesh_py import create_node + +note_doc = define_database_entry_type("mesh.note", [("body", 0)]) +node = create_node("mesh.cc", runtime_id="python-runtime") +node.register_document(note_doc) +node.put(note_doc, "note:1", {"body": "hello"}) +``` + +## Wire Parity + +The Python interop peer is `cultcache_py.interop`: + +```powershell +python -m cultcache_py.interop write --file cache.cc --runtime-id python +python -m cultcache_py.interop read --file cache.cc +``` + +`packages/cultcache-ts/test/cult-cache.test.ts` includes Python in the shared +CultCache v1 parity matrix with TypeScript, Rust, and C#. diff --git a/packages/cultcache-py/pyproject.toml b/packages/cultcache-py/pyproject.toml new file mode 100644 index 00000000..a654a831 --- /dev/null +++ b/packages/cultcache-py/pyproject.toml @@ -0,0 +1,28 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "cultcache-py" +version = "0.1.0" +description = "Python CultCache, CultNet, and CultMesh runtime package for GameCult typed state." +readme = "README.md" +requires-python = ">=3.10" +license = { text = "Private GameCult infrastructure" } +authors = [{ name = "GameCult" }] +classifiers = [ + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", +] +dependencies = ["msgpack>=1.0"] + +[project.optional-dependencies] +msgpack = ["msgpack>=1.0"] + +[project.scripts] +cultcache-py-interop = "cultcache_py.interop:main" + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/packages/cultcache-py/src/cultcache_py/__init__.py b/packages/cultcache-py/src/cultcache_py/__init__.py new file mode 100644 index 00000000..07721ef4 --- /dev/null +++ b/packages/cultcache-py/src/cultcache_py/__init__.py @@ -0,0 +1,29 @@ +from .backing_store import BackingStore, CultCacheEnvelope +from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember +from .cache import CultCache, CultCacheBuilder +from .documents import ( + DatabaseEntryField, + DocumentDefinition, + database_entry_field, + define_database_entry_type, + define_document_registry, + define_document_type, +) +from .stores import JsonLinesBackingStore, SingleFileMessagePackBackingStore + +__all__ = [ + "BackingStore", + "CultCache", + "CultCacheBuilder", + "CultCacheEnvelope", + "CultCacheSchemaCatalogEntry", + "CultCacheSchemaCatalogMember", + "DatabaseEntryField", + "DocumentDefinition", + "JsonLinesBackingStore", + "SingleFileMessagePackBackingStore", + "database_entry_field", + "define_database_entry_type", + "define_document_registry", + "define_document_type", +] diff --git a/packages/cultcache-py/src/cultcache_py/backing_store.py b/packages/cultcache-py/src/cultcache_py/backing_store.py new file mode 100644 index 00000000..5a21e496 --- /dev/null +++ b/packages/cultcache-py/src/cultcache_py/backing_store.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Protocol + + +@dataclass(frozen=True) +class CultCacheSchemaCatalogMember: + slot: int + member_name: str + type_name: str + is_reference: bool = False + is_many: bool = False + target_schema_name: str | None = None + is_name: bool = False + index_alias: str | None = None + + +@dataclass(frozen=True) +class CultCacheSchemaCatalogEntry: + schema_id: str + schema_name: str + schema_version: str + content_hash: str + canonical_schema_json: str + compatible_schema_ids: tuple[str, ...] = field(default_factory=tuple) + members: tuple[CultCacheSchemaCatalogMember, ...] = field(default_factory=tuple) + + +@dataclass(frozen=True) +class CultCacheEnvelope: + key: str + type: str + payload: bytes + stored_at: str + schema_id: str | None = None + catalog_entry: CultCacheSchemaCatalogEntry | None = None + + @classmethod + def create( + cls, + *, + key: str, + type: str, + payload: bytes, + schema_id: str | None = None, + catalog_entry: CultCacheSchemaCatalogEntry | None = None, + ) -> "CultCacheEnvelope": + return cls( + key=key, + type=type, + payload=payload, + stored_at=datetime.now(timezone.utc).isoformat(), + schema_id=schema_id, + catalog_entry=catalog_entry, + ) + + +class BackingStore(Protocol): + def pull_all(self) -> list[CultCacheEnvelope]: + ... + + def push(self, envelope: CultCacheEnvelope) -> None: + ... + + def delete(self, type: str, key: str) -> None: + ... + + def push_all(self, envelopes: list[CultCacheEnvelope]) -> None: + existing = { + (envelope.type, envelope.key): envelope for envelope in self.pull_all() + } + for envelope in envelopes: + existing[(envelope.type, envelope.key)] = envelope + self._replace_all(list(existing.values())) + + def _replace_all(self, envelopes: list[CultCacheEnvelope]) -> None: + ... diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py new file mode 100644 index 00000000..09392f0b --- /dev/null +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -0,0 +1,271 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Generic, TypeVar + +from .backing_store import BackingStore, CultCacheEnvelope +from .documents import DocumentDefinition, extract_value + +T = TypeVar("T") +GLOBAL_KEY = "__global__" + + +class CultCacheError(RuntimeError): + pass + + +@dataclass +class _State: + documents: dict[str, DocumentDefinition[Any]] = field(default_factory=dict) + values: dict[tuple[str, str], Any] = field(default_factory=dict) + envelopes: dict[tuple[str, str], CultCacheEnvelope] = field(default_factory=dict) + stores_by_type: dict[str, list[BackingStore]] = field(default_factory=dict) + generic_stores: list[BackingStore] = field(default_factory=list) + name_extractors: dict[str, str | Any] = field(default_factory=dict) + index_extractors: dict[str, dict[str, str | Any]] = field(default_factory=dict) + names: dict[tuple[str, str], str] = field(default_factory=dict) + indexes: dict[tuple[str, str, str], str] = field(default_factory=dict) + + +class CultCacheBuilder: + def __init__(self) -> None: + self._cache = CultCache() + + def register_document_type(self, document: DocumentDefinition[Any]) -> "CultCacheBuilder": + self._cache.register_document_type(document) + return self + + def register_registry(self, documents: list[DocumentDefinition[Any]] | tuple[DocumentDefinition[Any], ...]) -> "CultCacheBuilder": + self._cache.register_registry(documents) + return self + + def register_name_lookup(self, document: DocumentDefinition[Any], extractor: str | Any) -> "CultCacheBuilder": + self._cache.register_name_lookup(document, extractor) + return self + + def register_index(self, document: DocumentDefinition[Any], index: str, extractor: str | Any) -> "CultCacheBuilder": + self._cache.register_index(document, index, extractor) + return self + + def add_backing_store(self, store: BackingStore, types: list[str] | tuple[str, ...] | set[str]) -> "CultCacheBuilder": + self._cache.add_backing_store(store, types) + return self + + def add_generic_store(self, store: BackingStore) -> "CultCacheBuilder": + self._cache.add_generic_store(store) + return self + + def build(self) -> "CultCache": + return self._cache + + +class CultCache: + GLOBAL_KEY = GLOBAL_KEY + + @classmethod + def builder(cls) -> CultCacheBuilder: + return CultCacheBuilder() + + def __init__(self) -> None: + self._state = _State() + + def register_document_type(self, document: DocumentDefinition[Any]) -> None: + if document.type in self._state.documents: + raise CultCacheError(f"Document type already registered: {document.type}") + self._state.documents[document.type] = document + if document.name is not None: + self.register_name_lookup(document, document.name) + for index, extractor in document.indexes.items(): + self.register_index(document, index, extractor) + + def register_registry(self, documents: list[DocumentDefinition[Any]] | tuple[DocumentDefinition[Any], ...]) -> None: + for document in documents: + self.register_document_type(document) + + def register_name_lookup(self, document: DocumentDefinition[Any], extractor: str | Any) -> None: + self._assert_registered(document) + self._state.name_extractors[document.type] = extractor + self._rebuild_indexes() + + def register_index(self, document: DocumentDefinition[Any], index: str, extractor: str | Any) -> None: + self._assert_registered(document) + self._state.index_extractors.setdefault(document.type, {})[index] = extractor + self._rebuild_indexes() + + def add_backing_store(self, store: BackingStore, types: list[str] | tuple[str, ...] | set[str]) -> None: + for type in types: + self._state.stores_by_type.setdefault(type, []).append(store) + + def add_generic_store(self, store: BackingStore) -> None: + self._state.generic_stores.append(store) + + def pull_all_backing_stores(self) -> None: + self._state.values.clear() + self._state.envelopes.clear() + seen_globals: set[str] = set() + for store in [*self._all_specific_stores(), *self._state.generic_stores]: + for envelope in store.pull_all(): + document = self._state.documents.get(envelope.type) + if document is None: + raise CultCacheError(f"Unknown persisted document type: {envelope.type}") + if document.global_document: + if envelope.type in seen_globals and envelope.key == GLOBAL_KEY: + raise CultCacheError(f"Duplicate global document for type: {envelope.type}") + seen_globals.add(envelope.type) + value = document.decode_payload(envelope.payload) + self._state.values[(envelope.type, envelope.key)] = value + self._state.envelopes[(envelope.type, envelope.key)] = envelope + self._rebuild_indexes() + + def get(self, document: DocumentDefinition[T], key: str) -> T | None: + self._assert_registered(document) + return self._state.values.get((document.type, key)) + + def get_required(self, document: DocumentDefinition[T], key: str) -> T: + value = self.get(document, key) + if value is None: + raise CultCacheError(f"Missing {document.type}:{key}") + return value + + def get_all(self, document: DocumentDefinition[T]) -> list[T]: + self._assert_registered(document) + return [value for (type, _), value in self._state.values.items() if type == document.type] + + def get_envelope(self, document: DocumentDefinition[Any], key: str) -> CultCacheEnvelope | None: + self._assert_registered(document) + return self._state.envelopes.get((document.type, key)) + + def get_required_envelope(self, document: DocumentDefinition[Any], key: str) -> CultCacheEnvelope: + envelope = self.get_envelope(document, key) + if envelope is None: + raise CultCacheError(f"Missing envelope {document.type}:{key}") + return envelope + + def get_global(self, document: DocumentDefinition[T]) -> T | None: + self._assert_global(document) + return self.get(document, GLOBAL_KEY) + + def get_required_global(self, document: DocumentDefinition[T]) -> T: + self._assert_global(document) + return self.get_required(document, GLOBAL_KEY) + + def get_key_by_name(self, document: DocumentDefinition[Any], name: str) -> str | None: + self._assert_registered(document) + return self._state.names.get((document.type, name)) + + def get_by_name(self, document: DocumentDefinition[T], name: str) -> T | None: + key = self.get_key_by_name(document, name) + return None if key is None else self.get(document, key) + + def get_key_by_index(self, document: DocumentDefinition[Any], index: str, value: str) -> str | None: + self._assert_registered(document) + return self._state.indexes.get((document.type, index, value)) + + def get_by_index(self, document: DocumentDefinition[T], index: str, value: str) -> T | None: + key = self.get_key_by_index(document, index, value) + return None if key is None else self.get(document, key) + + def put(self, document: DocumentDefinition[T], key: str, value: T) -> None: + self._assert_registered(document) + if document.global_document and key != GLOBAL_KEY: + raise CultCacheError(f"Global document {document.type} must use key {GLOBAL_KEY}") + catalog_entry = document.catalog_entry() + envelope = CultCacheEnvelope.create( + key=key, + type=document.type, + payload=document.encode_payload(value), + schema_id=catalog_entry.schema_id, + catalog_entry=catalog_entry, + ) + stores = self._stores_for_type(document.type) + if not stores: + raise CultCacheError(f"No backing store registered for {document.type}") + for store in stores: + store.push(envelope) + self._state.values[(document.type, key)] = value + self._state.envelopes[(document.type, key)] = envelope + self._rebuild_indexes() + + def put_envelope(self, document: DocumentDefinition[T], envelope: CultCacheEnvelope) -> T: + self._assert_registered(document) + if envelope.type != document.type: + raise CultCacheError( + f"Envelope type {envelope.type} does not match document type {document.type}" + ) + value = document.decode_payload(envelope.payload) + for store in self._stores_for_type(document.type): + store.push(envelope) + self._state.values[(document.type, envelope.key)] = value + self._state.envelopes[(document.type, envelope.key)] = envelope + self._rebuild_indexes() + return value + + def put_global(self, document: DocumentDefinition[T], value: T) -> None: + self._assert_global(document) + self.put(document, GLOBAL_KEY, value) + + def update(self, document: DocumentDefinition[T], key: str, updater: Any) -> T: + current = self.get_required(document, key) + updated = updater(current) + self.put(document, key, updated) + return updated + + def update_global(self, document: DocumentDefinition[T], updater: Any) -> T: + self._assert_global(document) + return self.update(document, GLOBAL_KEY, updater) + + def delete(self, document: DocumentDefinition[Any], key: str) -> None: + self._assert_registered(document) + for store in self._stores_for_type(document.type): + store.delete(document.type, key) + self._state.values.pop((document.type, key), None) + self._state.envelopes.pop((document.type, key), None) + self._rebuild_indexes() + + def delete_global(self, document: DocumentDefinition[Any]) -> None: + self._assert_global(document) + self.delete(document, GLOBAL_KEY) + + def snapshot(self) -> dict[str, dict[str, Any]]: + out: dict[str, dict[str, Any]] = {} + for (type, key), value in self._state.values.items(): + out.setdefault(type, {})[key] = value + return out + + def _stores_for_type(self, type: str) -> list[BackingStore]: + specific = self._state.stores_by_type.get(type) + return specific if specific else self._state.generic_stores + + def _all_specific_stores(self) -> list[BackingStore]: + stores: list[BackingStore] = [] + seen: set[int] = set() + for routed in self._state.stores_by_type.values(): + for store in routed: + marker = id(store) + if marker not in seen: + stores.append(store) + seen.add(marker) + return stores + + def _rebuild_indexes(self) -> None: + self._state.names.clear() + self._state.indexes.clear() + for (type, key), value in self._state.values.items(): + name_extractor = self._state.name_extractors.get(type) + if name_extractor is not None: + name = extract_value(value, name_extractor) + if name is not None: + self._state.names[(type, str(name))] = key + for index, extractor in self._state.index_extractors.get(type, {}).items(): + index_value = extract_value(value, extractor) + if index_value is not None: + self._state.indexes[(type, index, str(index_value))] = key + + def _assert_registered(self, document: DocumentDefinition[Any]) -> None: + if self._state.documents.get(document.type) is not document: + raise CultCacheError(f"Document type is not registered on this cache: {document.type}") + + def _assert_global(self, document: DocumentDefinition[Any]) -> None: + self._assert_registered(document) + if not document.global_document: + raise CultCacheError(f"Document type is not global: {document.type}") diff --git a/packages/cultcache-py/src/cultcache_py/documents.py b/packages/cultcache-py/src/cultcache_py/documents.py new file mode 100644 index 00000000..10cf3d80 --- /dev/null +++ b/packages/cultcache-py/src/cultcache_py/documents.py @@ -0,0 +1,298 @@ +from __future__ import annotations + +import json +from dataclasses import MISSING, dataclass, field +from typing import Any, Callable, Generic, Iterable, TypeVar + +from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember + +T = TypeVar("T") + +Encoder = Callable[[T], Any] +Decoder = Callable[[Any], T] +Validator = Callable[[T], None] +Extractor = Callable[[T], str | int | float | bool | None] + + +def _default_encode(value: Any) -> Any: + if hasattr(value, "model_dump"): + return value.model_dump() + if hasattr(value, "to_dict"): + return value.to_dict() + if hasattr(value, "__dict__"): + return dict(value.__dict__) + return value + + +def _default_decode(value: Any) -> Any: + return value + + +def _json_payload_encoder(value: Any) -> bytes: + return json.dumps(value, ensure_ascii=False, sort_keys=True).encode("utf-8") + + +def _json_payload_decoder(payload: bytes) -> Any: + return json.loads(payload.decode("utf-8")) + + +def _messagepack_payload_encoder(value: Any) -> bytes: + try: + import msgpack # type: ignore + except ModuleNotFoundError as exc: + raise RuntimeError( + "MessagePack payload encoding requires the optional 'msgpack' dependency. " + "Install with: python -m pip install cultcache-py[msgpack]" + ) from exc + return msgpack.packb(value, use_bin_type=True) + + +def _messagepack_payload_decoder(payload: bytes) -> Any: + try: + import msgpack # type: ignore + except ModuleNotFoundError as exc: + raise RuntimeError( + "MessagePack payload decoding requires the optional 'msgpack' dependency. " + "Install with: python -m pip install cultcache-py[msgpack]" + ) from exc + return msgpack.unpackb(payload, raw=False) + + +@dataclass(frozen=True) +class DatabaseEntryField: + name: str + key: int + default: Any = MISSING + + +def database_entry_field(name: str, key: int, *, default: Any = MISSING) -> DatabaseEntryField: + if key < 0: + raise ValueError("DatabaseEntry field keys must be non-negative integers") + return DatabaseEntryField(name=name, key=key, default=default) + + +@dataclass(frozen=True) +class DocumentDefinition(Generic[T]): + type: str + encode: Encoder[T] = _default_encode + decode: Decoder[T] = _default_decode + validate: Validator[T] | None = None + global_document: bool = False + name: str | Extractor[T] | None = None + indexes: dict[str, str | Extractor[T]] = field(default_factory=dict) + payload_encoder: Callable[[Any], bytes] = _json_payload_encoder + payload_decoder: Callable[[bytes], Any] = _json_payload_decoder + schema_id: str | None = None + schema_name: str | None = None + schema_version: str | None = None + content_hash: str | None = None + canonical_schema_json: str | None = None + compatible_schema_ids: tuple[str, ...] = field(default_factory=tuple) + members: tuple[CultCacheSchemaCatalogMember, ...] = field(default_factory=tuple) + + def encode_payload(self, value: T) -> bytes: + if self.validate: + self.validate(value) + return self.payload_encoder(self.encode(value)) + + def decode_payload(self, payload: bytes) -> T: + value = self.decode(self.payload_decoder(payload)) + if self.validate: + self.validate(value) + return value + + def catalog_entry(self) -> CultCacheSchemaCatalogEntry: + 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" + canonical_schema_json = self.canonical_schema_json or json.dumps( + { + "schemaName": schema_name, + "schemaVersion": schema_version, + "members": [ + { + "slot": member.slot, + "name": member.member_name, + "type": member.type_name, + "isReference": member.is_reference, + "many": member.is_many, + "targetSchemaName": member.target_schema_name, + "indexAlias": member.index_alias, + "isName": member.is_name, + } + for member in self.members + ], + }, + ensure_ascii=False, + separators=(",", ":"), + ) + compatible_schema_ids = self.compatible_schema_ids or (schema_id,) + return CultCacheSchemaCatalogEntry( + schema_id=schema_id, + schema_name=schema_name, + schema_version=schema_version, + content_hash=self.content_hash or schema_id, + canonical_schema_json=canonical_schema_json, + compatible_schema_ids=tuple(compatible_schema_ids), + members=tuple(self.members), + ) + + +def define_document_type( + type: str, + *, + encode: Encoder[T] = _default_encode, + decode: Decoder[T] = _default_decode, + validate: Validator[T] | None = None, + global_document: bool = False, + name: str | Extractor[T] | None = None, + indexes: dict[str, str | Extractor[T]] | None = None, + payload_encoder: Callable[[Any], bytes] = _json_payload_encoder, + payload_decoder: Callable[[bytes], Any] = _json_payload_decoder, + schema_id: str | None = None, + schema_name: str | None = None, + schema_version: str | None = None, + content_hash: str | None = None, + canonical_schema_json: str | None = None, + compatible_schema_ids: Iterable[str] | None = None, + members: Iterable[CultCacheSchemaCatalogMember] | None = None, +) -> DocumentDefinition[T]: + return DocumentDefinition( + type=type, + encode=encode, + decode=decode, + validate=validate, + global_document=global_document, + name=name, + indexes=indexes or {}, + payload_encoder=payload_encoder, + payload_decoder=payload_decoder, + schema_id=schema_id, + schema_name=schema_name, + schema_version=schema_version, + content_hash=content_hash, + canonical_schema_json=canonical_schema_json, + compatible_schema_ids=tuple(compatible_schema_ids or ()), + members=tuple(members or ()), + ) + + +def define_database_entry_type( + type: str, + fields: Iterable[DatabaseEntryField | tuple[str, int] | tuple[str, int, Any]], + *, + cls: type[T] | None = None, + validate: Validator[T] | None = None, + global_document: bool = False, + name: str | Extractor[T] | None = None, + indexes: dict[str, str | Extractor[T]] | None = None, + schema_id: str | None = None, + schema_name: str | None = None, + schema_version: str | None = None, + content_hash: str | None = None, + canonical_schema_json: str | None = None, + compatible_schema_ids: Iterable[str] | None = None, +) -> DocumentDefinition[T]: + """Define a Rust/C#-style DatabaseEntry payload formatter. + + Payloads are MessagePack arrays indexed by explicit field keys. Missing + slots are written as nil, so deleted fields can leave their slot reserved + 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) + member_name_lookup = name if isinstance(name, str) else None + index_lookup = indexes or {} + members = tuple( + CultCacheSchemaCatalogMember( + slot=field.key, + member_name=_pascal_case(field.name), + type_name="System.Object", + is_name=field.name == member_name_lookup, + index_alias=next( + (alias for alias, extractor in index_lookup.items() if extractor == field.name), + None, + ), + ) + 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) + + return define_document_type( + type, + encode=encode, + decode=decode, + validate=validate, + global_document=global_document, + name=name, + indexes=indexes, + payload_encoder=_messagepack_payload_encoder, + payload_decoder=_messagepack_payload_decoder, + schema_id=schema_id, + schema_name=schema_name, + schema_version=schema_version, + content_hash=content_hash, + canonical_schema_json=canonical_schema_json, + compatible_schema_ids=compatible_schema_ids, + members=members, + ) + + +def define_document_registry(*documents: DocumentDefinition[Any]) -> tuple[DocumentDefinition[Any], ...]: + return tuple(documents) + + +def extract_value(value: Any, extractor: str | Extractor[Any]) -> str | int | float | bool | None: + if callable(extractor): + return extractor(value) + if isinstance(value, dict): + return value.get(extractor) + return getattr(value, extractor, None) + + +def _normalize_database_entry_field( + field: DatabaseEntryField | tuple[str, int] | tuple[str, int, Any], +) -> DatabaseEntryField: + if isinstance(field, DatabaseEntryField): + return field + if len(field) == 2: + return database_entry_field(field[0], field[1]) + if len(field) == 3: + return database_entry_field(field[0], field[1], default=field[2]) + raise ValueError("DatabaseEntry field tuples must be (name, key) or (name, key, default)") + + +def _pascal_case(name: str) -> str: + return "".join(part[:1].upper() + part[1:] for part in name.split("_") if part) diff --git a/packages/cultcache-py/src/cultcache_py/interop.py b/packages/cultcache-py/src/cultcache_py/interop.py new file mode 100644 index 00000000..021969f1 --- /dev/null +++ b/packages/cultcache-py/src/cultcache_py/interop.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import argparse +import json +from typing import Any + +import msgpack + +from . import CultCache, CultCacheSchemaCatalogMember, SingleFileMessagePackBackingStore, define_document_type + +INTEROP_SCHEMA_VERSION = "cultcache.interop_note.v1" + +interop_note_document = define_document_type( + "cultcache.interop-note", + encode=lambda note: [ + note["schemaVersion"], + note["documentId"], + note["authorRuntimeId"], + note["title"], + note["body"], + note["tags"], + ], + decode=lambda slots: { + "schemaVersion": slots[0], + "documentId": slots[1], + "authorRuntimeId": slots[2], + "title": slots[3], + "body": slots[4], + "tags": slots[5] if len(slots) > 5 and slots[5] is not None else [], + }, + name="documentId", + payload_encoder=lambda value: msgpack.packb(value, use_bin_type=True), + payload_decoder=lambda payload: msgpack.unpackb(payload, raw=False), + schema_id="cultcache.interop-note", + schema_name="cultcache.interop-note", + schema_version=INTEROP_SCHEMA_VERSION, + content_hash="cultcache.interop-note", + canonical_schema_json='{"schemaName":"cultcache.interop-note","schemaVersion":"cultcache.interop_note.v1","members":[{"slot":0,"name":"SchemaVersion","type":"System.String","isReference":false,"many":false,"targetSchemaName":null,"indexAlias":null,"isName":false},{"slot":1,"name":"DocumentId","type":"System.String","isReference":false,"many":false,"targetSchemaName":null,"indexAlias":null,"isName":true},{"slot":2,"name":"AuthorRuntimeId","type":"System.String","isReference":false,"many":false,"targetSchemaName":null,"indexAlias":null,"isName":false},{"slot":3,"name":"Title","type":"System.String","isReference":false,"many":false,"targetSchemaName":null,"indexAlias":null,"isName":false},{"slot":4,"name":"Body","type":"System.String","isReference":false,"many":false,"targetSchemaName":null,"indexAlias":null,"isName":false},{"slot":5,"name":"Tags","type":"System.String[]","isReference":false,"many":false,"targetSchemaName":null,"indexAlias":null,"isName":false}]}', + compatible_schema_ids=("cultcache.interop-note",), + members=( + CultCacheSchemaCatalogMember(0, "SchemaVersion", "System.String"), + CultCacheSchemaCatalogMember(1, "DocumentId", "System.String", is_name=True), + CultCacheSchemaCatalogMember(2, "AuthorRuntimeId", "System.String"), + CultCacheSchemaCatalogMember(3, "Title", "System.String"), + CultCacheSchemaCatalogMember(4, "Body", "System.String"), + CultCacheSchemaCatalogMember(5, "Tags", "System.String[]"), + ), +) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="cultcache-py-interop") + parser.add_argument("mode", choices=("write", "read")) + parser.add_argument("--file", required=True) + parser.add_argument("--runtime-id") + args = parser.parse_args(argv) + + if args.mode == "write": + if not args.runtime_id: + parser.error("write requires --runtime-id") + note = write_note(args.file, args.runtime_id) + else: + note = read_note(args.file) + print(json.dumps(note, separators=(",", ":"))) + return 0 + + +def write_note(file: str, runtime_id: str) -> dict[str, Any]: + cache = build_cache(file) + cache.pull_all_backing_stores() + note = { + "schemaVersion": INTEROP_SCHEMA_VERSION, + "documentId": f"note:{runtime_id}", + "authorRuntimeId": runtime_id, + "title": f"{runtime_id} wrote a CultCache note", + "body": "The v1 store format is the contract.", + "tags": [runtime_id, "python", "interop"], + } + cache.put(interop_note_document, note["documentId"], note) + return note + + +def read_note(file: str) -> dict[str, Any]: + cache = build_cache(file) + cache.pull_all_backing_stores() + notes = cache.get_all(interop_note_document) + if not notes: + raise RuntimeError("No cultcache.interop-note records found.") + return notes[0] + + +def build_cache(file: str) -> CultCache: + return ( + CultCache.builder() + .register_document_type(interop_note_document) + .add_generic_store(SingleFileMessagePackBackingStore(file)) + .build() + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py new file mode 100644 index 00000000..3587e717 --- /dev/null +++ b/packages/cultcache-py/src/cultcache_py/stores.py @@ -0,0 +1,318 @@ +from __future__ import annotations + +import base64 +import json +import os +import threading +from pathlib import Path +from typing import Any + +from .backing_store import ( + CultCacheEnvelope, + CultCacheSchemaCatalogEntry, + CultCacheSchemaCatalogMember, +) + +STORE_FORMAT_VERSION = "cultcache.store.v1" + + +class JsonLinesBackingStore: + def __init__(self, path: str | os.PathLike[str]) -> None: + self.path = Path(path) + self._lock = threading.RLock() + + def pull_all(self) -> list[CultCacheEnvelope]: + with self._lock: + if not self.path.exists(): + return [] + envelopes: list[CultCacheEnvelope] = [] + for line_number, line in enumerate(self.path.read_text(encoding="utf-8").splitlines(), start=1): + if not line.strip(): + continue + raw = json.loads(line) + try: + payload = base64.b64decode(raw["payload"]) + envelopes.append( + CultCacheEnvelope( + key=raw["key"], + type=raw["type"], + payload=payload, + stored_at=raw.get("stored_at", raw.get("storedAt")), + schema_id=raw.get("schema_id", raw.get("schemaId")), + ) + ) + except KeyError as exc: + raise ValueError(f"Malformed CultCache JSONL envelope at {self.path}:{line_number}: missing {exc}") from exc + return envelopes + + def push(self, envelope: CultCacheEnvelope) -> None: + with self._lock: + existing = {(item.type, item.key): item for item in self.pull_all()} + existing[(envelope.type, envelope.key)] = envelope + self._replace_all(list(existing.values())) + + def push_all(self, envelopes: list[CultCacheEnvelope]) -> None: + with self._lock: + existing = {(item.type, item.key): item for item in self.pull_all()} + for envelope in envelopes: + existing[(envelope.type, envelope.key)] = envelope + self._replace_all(list(existing.values())) + + def delete(self, type: str, key: str) -> None: + with self._lock: + existing = [item for item in self.pull_all() if not (item.type == type and item.key == key)] + self._replace_all(existing) + + def _replace_all(self, envelopes: list[CultCacheEnvelope]) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + temp = self.path.with_suffix(self.path.suffix + ".tmp") + lines = [] + for envelope in sorted(envelopes, key=lambda item: (item.type, item.key)): + lines.append(json.dumps({ + "key": envelope.key, + "type": envelope.type, + "payload": base64.b64encode(envelope.payload).decode("ascii"), + "stored_at": envelope.stored_at, + }, ensure_ascii=False, sort_keys=True)) + temp.write_text("\n".join(lines) + ("\n" if lines else ""), encoding="utf-8") + temp.replace(self.path) + + +class SingleFileMessagePackBackingStore: + def __init__(self, path: str | os.PathLike[str]) -> None: + self.path = Path(path) + self._lock = threading.RLock() + + def pull_all(self) -> list[CultCacheEnvelope]: + msgpack = self._msgpack() + with self._lock: + if not self.path.exists(): + return [] + data = self.path.read_bytes() + if not data: + return [] + decoded = msgpack.unpackb(data, raw=False) + snapshot = _decode_v1_snapshot(decoded) + if snapshot is not None: + return snapshot + return _decode_legacy_envelopes(decoded, msgpack) + + def push(self, envelope: CultCacheEnvelope) -> None: + with self._lock: + existing = {(item.type, item.key): item for item in self.pull_all()} + existing[(envelope.type, envelope.key)] = envelope + self._replace_all(list(existing.values())) + + def push_all(self, envelopes: list[CultCacheEnvelope]) -> None: + with self._lock: + existing = {(item.type, item.key): item for item in self.pull_all()} + for envelope in envelopes: + existing[(envelope.type, envelope.key)] = envelope + self._replace_all(list(existing.values())) + + def delete(self, type: str, key: str) -> None: + with self._lock: + existing = [item for item in self.pull_all() if not (item.type == type and item.key == key)] + self._replace_all(existing) + + def _replace_all(self, envelopes: list[CultCacheEnvelope]) -> None: + msgpack = self._msgpack() + self.path.parent.mkdir(parents=True, exist_ok=True) + temp = self.path.with_suffix(self.path.suffix + ".tmp") + temp.write_bytes(msgpack.packb(_encode_v1_snapshot(envelopes), use_bin_type=True)) + temp.replace(self.path) + + @staticmethod + def _msgpack() -> Any: + try: + import msgpack # type: ignore + except ModuleNotFoundError as exc: + raise RuntimeError( + "SingleFileMessagePackBackingStore requires the optional 'msgpack' dependency. " + "Install with: python -m pip install cultcache-py[msgpack]" + ) from exc + return msgpack + + +def _encode_v1_snapshot(envelopes: list[CultCacheEnvelope]) -> list[Any]: + catalog_by_schema_id: dict[str, CultCacheSchemaCatalogEntry] = {} + for envelope in envelopes: + schema_id = _schema_id_for(envelope) + if schema_id not in catalog_by_schema_id: + catalog_by_schema_id[schema_id] = envelope.catalog_entry or _default_catalog_entry(envelope) + + catalog = [ + _encode_catalog_entry(entry) + for entry in sorted(catalog_by_schema_id.values(), key=lambda item: item.schema_name) + ] + records = [ + [envelope.key, _schema_id_for(envelope), envelope.stored_at, envelope.payload] + for envelope in sorted(envelopes, key=lambda item: item.key) + ] + return [STORE_FORMAT_VERSION, catalog, records] + + +def _decode_v1_snapshot(decoded: Any) -> list[CultCacheEnvelope] | None: + if not isinstance(decoded, list) or not decoded or decoded[0] != STORE_FORMAT_VERSION: + return None + if len(decoded) < 3 or not isinstance(decoded[1], list) or not isinstance(decoded[2], list): + raise ValueError("CultCache v1 snapshot must contain a schema catalog and record array") + + catalog_by_schema_id: dict[str, CultCacheSchemaCatalogEntry] = {} + for raw_entry in decoded[1]: + entry = _decode_catalog_entry(raw_entry) + catalog_by_schema_id[entry.schema_id] = entry + for compatible_schema_id in entry.compatible_schema_ids: + catalog_by_schema_id[compatible_schema_id] = entry + + envelopes: list[CultCacheEnvelope] = [] + for raw_record in decoded[2]: + if not isinstance(raw_record, list) or len(raw_record) < 4: + raise ValueError("CultCache persisted records must be MessagePack arrays") + key, schema_id, stored_at, payload = raw_record[:4] + if not isinstance(key, str) or not key: + raise ValueError("CultCache persisted records must declare a key") + if not isinstance(schema_id, str) or not schema_id: + raise ValueError("CultCache persisted records must declare a schema id") + if not isinstance(stored_at, str) or not stored_at: + raise ValueError("CultCache persisted records must declare storedAt") + catalog_entry = catalog_by_schema_id.get(schema_id) + if catalog_entry is None: + raise ValueError(f'CultCache persisted record "{key}" references missing schema id "{schema_id}"') + envelopes.append( + CultCacheEnvelope( + key=key, + type=catalog_entry.schema_name, + payload=_normalize_payload(payload), + stored_at=stored_at, + schema_id=schema_id, + catalog_entry=catalog_entry, + ) + ) + return envelopes + + +def _decode_legacy_envelopes(decoded: Any, msgpack: Any) -> list[CultCacheEnvelope]: + if not isinstance(decoded, list): + raise ValueError("CultCache MessagePack store is not a recognized snapshot") + envelopes: list[CultCacheEnvelope] = [] + for item in decoded: + if not isinstance(item, dict): + raise ValueError("CultCache legacy envelopes must be MessagePack maps") + try: + payload = item["payload"] + envelopes.append( + CultCacheEnvelope( + key=item["key"], + type=item["type"], + payload=payload if isinstance(payload, bytes) else msgpack.packb(payload, use_bin_type=True), + stored_at=item.get("storedAt", item.get("stored_at")), + schema_id=item.get("schemaId", item.get("schema_id")), + ) + ) + except KeyError as exc: + raise ValueError(f"Malformed CultCache legacy envelope: missing {exc}") from exc + return envelopes + + +def _encode_catalog_entry(entry: CultCacheSchemaCatalogEntry) -> list[Any]: + return [ + entry.schema_id, + entry.schema_name, + entry.schema_version, + entry.content_hash, + entry.canonical_schema_json, + list(entry.compatible_schema_ids or (entry.schema_id,)), + [_encode_catalog_member(member) for member in entry.members], + ] + + +def _decode_catalog_entry(value: Any) -> CultCacheSchemaCatalogEntry: + if not isinstance(value, list): + raise ValueError("CultCache schema catalog entries must be MessagePack arrays") + schema_id = value[0] if len(value) > 0 else "" + schema_name = value[1] if len(value) > 1 else "" + schema_version = value[2] if len(value) > 2 else "" + content_hash = value[3] if len(value) > 3 else "" + canonical_schema_json = value[4] if len(value) > 4 else "" + compatible_schema_ids = value[5] if len(value) > 5 else [] + members = value[6] if len(value) > 6 else [] + if not isinstance(schema_id, str) or not schema_id or not isinstance(schema_name, str) or not schema_name: + raise ValueError("CultCache schema catalog entries must declare schemaId and schemaName") + if not isinstance(compatible_schema_ids, list) or not all(isinstance(item, str) and item for item in compatible_schema_ids): + raise ValueError(f'CultCache schema catalog entry "{schema_id}" has invalid compatible schema ids') + if not isinstance(members, list): + raise ValueError(f'CultCache schema catalog entry "{schema_id}" has invalid members') + return CultCacheSchemaCatalogEntry( + schema_id=schema_id, + schema_name=schema_name, + schema_version=schema_version if isinstance(schema_version, str) and schema_version else f"{schema_name}.v1", + content_hash=content_hash if isinstance(content_hash, str) and content_hash else schema_id, + canonical_schema_json=canonical_schema_json if isinstance(canonical_schema_json, str) else "", + compatible_schema_ids=tuple(compatible_schema_ids), + members=tuple(_decode_catalog_member(member) for member in members), + ) + + +def _encode_catalog_member(member: CultCacheSchemaCatalogMember) -> list[Any]: + return [ + member.slot, + member.member_name, + member.type_name, + member.is_reference, + member.is_many, + member.target_schema_name, + member.is_name, + member.index_alias, + ] + + +def _decode_catalog_member(value: Any) -> CultCacheSchemaCatalogMember: + if not isinstance(value, list): + raise ValueError("CultCache schema catalog members must be MessagePack arrays") + slot = value[0] if len(value) > 0 else -1 + member_name = value[1] if len(value) > 1 else "" + type_name = value[2] if len(value) > 2 else "" + if not isinstance(slot, int) or slot < 0 or not isinstance(member_name, str) or not member_name: + raise ValueError("CultCache schema catalog member has invalid slot or name") + if not isinstance(type_name, str) or not type_name: + raise ValueError("CultCache schema catalog member has invalid type") + return CultCacheSchemaCatalogMember( + slot=slot, + member_name=member_name, + type_name=type_name, + is_reference=(value[3] if len(value) > 3 else False) is True, + is_many=(value[4] if len(value) > 4 else False) is True, + target_schema_name=value[5] if len(value) > 5 and isinstance(value[5], str) else None, + is_name=(value[6] if len(value) > 6 else False) is True, + index_alias=value[7] if len(value) > 7 and isinstance(value[7], str) else None, + ) + + +def _schema_id_for(envelope: CultCacheEnvelope) -> str: + return envelope.schema_id or envelope.type + + +def _default_catalog_entry(envelope: CultCacheEnvelope) -> CultCacheSchemaCatalogEntry: + schema_id = _schema_id_for(envelope) + return CultCacheSchemaCatalogEntry( + schema_id=schema_id, + schema_name=envelope.type, + schema_version=f"{envelope.type}.v1", + content_hash=schema_id, + canonical_schema_json=json.dumps( + {"schemaName": envelope.type, "schemaVersion": f"{envelope.type}.v1", "members": []}, + separators=(",", ":"), + ), + compatible_schema_ids=(schema_id,), + ) + + +def _normalize_payload(payload: Any) -> bytes: + if isinstance(payload, bytes): + return payload + if isinstance(payload, bytearray): + return bytes(payload) + if isinstance(payload, list) and all(isinstance(item, int) and 0 <= item <= 255 for item in payload): + return bytes(payload) + raise ValueError("CultCache record payload must be binary MessagePack bytes") diff --git a/packages/cultcache-py/src/cultmesh_py/__init__.py b/packages/cultcache-py/src/cultmesh_py/__init__.py new file mode 100644 index 00000000..89e94f5e --- /dev/null +++ b/packages/cultcache-py/src/cultmesh_py/__init__.py @@ -0,0 +1,3 @@ +from .node import CultMeshNode, create_node + +__all__ = ["CultMeshNode", "create_node"] diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py new file mode 100644 index 00000000..40fc1c5b --- /dev/null +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from cultcache_py import CultCache, SingleFileMessagePackBackingStore +from cultcache_py.documents import DocumentDefinition + + +@dataclass +class CultMeshNode: + cache: CultCache = field(default_factory=CultCache) + runtime_id: str = "python-runtime" + + def register_document(self, document: DocumentDefinition[Any]) -> None: + self.cache.register_document_type(document) + + def put(self, document: DocumentDefinition[Any], key: str, value: Any) -> None: + self.cache.put(document, key, value) + + def get(self, document: DocumentDefinition[Any], key: str) -> Any: + return self.cache.get(document, key) + + def pull(self) -> None: + self.cache.pull_all_backing_stores() + + +def create_node(cache_path: str | Path | None = None, *, runtime_id: str = "python-runtime") -> CultMeshNode: + cache = CultCache() + if cache_path is not None: + cache.add_generic_store(SingleFileMessagePackBackingStore(cache_path)) + return CultMeshNode(cache=cache, runtime_id=runtime_id) diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py new file mode 100644 index 00000000..fad9f8ac --- /dev/null +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -0,0 +1,24 @@ +from .framing import decode_frame, encode_frame, read_frame, write_frame +from .messages import ( + CultNetMessage, + document_delete, + document_put_raw, + hello, + parse_message, + schema_catalog_request, + snapshot_request, +) + +__all__ = [ + "CultNetMessage", + "decode_frame", + "document_delete", + "document_put_raw", + "encode_frame", + "hello", + "parse_message", + "read_frame", + "schema_catalog_request", + "snapshot_request", + "write_frame", +] diff --git a/packages/cultcache-py/src/cultnet_py/framing.py b/packages/cultcache-py/src/cultnet_py/framing.py new file mode 100644 index 00000000..33ad92f0 --- /dev/null +++ b/packages/cultcache-py/src/cultnet_py/framing.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from typing import BinaryIO + + +def encode_frame(payload: bytes) -> bytes: + if len(payload) > 0xFFFFFFFF: + raise ValueError("CultNet frames cannot exceed 4GiB") + return len(payload).to_bytes(4, "big") + payload + + +def decode_frame(frame: bytes) -> bytes: + if len(frame) < 4: + raise ValueError("CultNet frame is shorter than the 4-byte length prefix") + length = int.from_bytes(frame[:4], "big") + payload = frame[4:] + if len(payload) != length: + raise ValueError(f"CultNet frame declared {length} bytes but carried {len(payload)}") + return payload + + +def read_frame(stream: BinaryIO) -> bytes: + prefix = stream.read(4) + if prefix == b"": + raise EOFError("No CultNet frame prefix available") + if len(prefix) != 4: + raise EOFError("Partial CultNet frame prefix") + length = int.from_bytes(prefix, "big") + payload = stream.read(length) + if len(payload) != length: + raise EOFError("Partial CultNet frame payload") + return payload + + +def write_frame(stream: BinaryIO, payload: bytes) -> None: + stream.write(encode_frame(payload)) diff --git a/packages/cultcache-py/src/cultnet_py/messages.py b/packages/cultcache-py/src/cultnet_py/messages.py new file mode 100644 index 00000000..ab4ffa1a --- /dev/null +++ b/packages/cultcache-py/src/cultnet_py/messages.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import msgpack + + +@dataclass(frozen=True) +class CultNetMessage: + schema_version: str + body: dict[str, Any] + + def to_wire(self) -> dict[str, Any]: + return {"schemaVersion": self.schema_version, **self.body} + + def to_bytes(self) -> bytes: + return msgpack.packb(self.to_wire(), use_bin_type=True) + + +def parse_message(payload: bytes) -> CultNetMessage: + decoded = msgpack.unpackb(payload, raw=False) + if not isinstance(decoded, dict): + raise ValueError("CultNet schema-v0 messages must be MessagePack maps") + schema_version = decoded.get("schemaVersion") + if not isinstance(schema_version, str) or not schema_version: + raise ValueError("CultNet schema-v0 messages must declare schemaVersion") + body = dict(decoded) + del body["schemaVersion"] + return CultNetMessage(schema_version=schema_version, body=body) + + +def hello( + *, + runtime_id: str, + supported_wire_contracts: list[str] | None = None, + supported_schema_versions: list[str] | None = None, + supported_mutation_contracts: list[dict[str, Any]] | None = None, +) -> CultNetMessage: + return CultNetMessage( + "cultnet.hello.v0", + { + "runtimeId": runtime_id, + "supportedWireContracts": supported_wire_contracts or ["cultnet.schema.v0"], + "supportedSchemaVersions": supported_schema_versions or [], + "supportedMutationContracts": supported_mutation_contracts or [], + }, + ) + + +def document_put_raw( + *, + key: str, + schema_id: str, + stored_at: str, + payload: bytes, + source_runtime_id: str | None = None, +) -> CultNetMessage: + record: dict[str, Any] = { + "schemaId": schema_id, + "recordKey": key, + "storedAt": stored_at, + "payloadEncoding": "messagepack", + "payload": payload, + } + if source_runtime_id: + record["sourceRuntimeId"] = source_runtime_id + return CultNetMessage("cultnet.document_put_raw.v0", {"record": record}) + + +def document_delete(*, schema_id: str, record_key: str, deleted_at: str) -> CultNetMessage: + return CultNetMessage( + "cultnet.document_delete.v0", + {"schemaId": schema_id, "recordKey": record_key, "deletedAt": deleted_at}, + ) + + +def snapshot_request(*, schema_ids: list[str] | None = None) -> CultNetMessage: + return CultNetMessage("cultnet.snapshot_request.v0", {"schemaIds": schema_ids or []}) + + +def schema_catalog_request(*, include_schema_json: bool = False) -> CultNetMessage: + return CultNetMessage( + "cultnet.schema_catalog_request.v0", + {"includeSchemaJson": include_schema_json}, + ) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py new file mode 100644 index 00000000..14d5bdfb --- /dev/null +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +import tempfile +import unittest +from dataclasses import asdict, dataclass +from pathlib import Path + +from cultcache_py import ( + CultCache, + JsonLinesBackingStore, + SingleFileMessagePackBackingStore, + define_database_entry_type, + define_document_type, +) +from cultcache_py.interop import read_note, write_note +from cultnet_py import decode_frame, encode_frame, hello, parse_message +from cultmesh_py import create_node + + +@dataclass +class Item: + name: str + category: str + value: int + + +def item_doc(): + return define_document_type( + "item", + encode=lambda item: asdict(item), + decode=lambda raw: Item(**raw), + name="name", + indexes={"category": "category"}, + ) + + +class CultCacheTests(unittest.TestCase): + def test_round_trips_registered_documents(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + document = item_doc() + cache = ( + CultCache.builder() + .register_document_type(document) + .add_generic_store(JsonLinesBackingStore(Path(tmp) / "cache.jsonl")) + .build() + ) + cache.pull_all_backing_stores() + cache.put(document, "item:potion", Item(name="Potion", category="Consumable", value=50)) + + loaded = ( + CultCache.builder() + .register_document_type(document) + .add_generic_store(JsonLinesBackingStore(Path(tmp) / "cache.jsonl")) + .build() + ) + loaded.pull_all_backing_stores() + + self.assertEqual(loaded.get_required(document, "item:potion").value, 50) + self.assertEqual(loaded.get_key_by_name(document, "Potion"), "item:potion") + self.assertEqual(loaded.get_by_index(document, "category", "Consumable").name, "Potion") + + def test_global_document_uses_singleton_key(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + settings = define_document_type("settings", global_document=True) + cache = ( + CultCache.builder() + .register_document_type(settings) + .add_generic_store(JsonLinesBackingStore(Path(tmp) / "cache.jsonl")) + .build() + ) + cache.pull_all_backing_stores() + cache.put_global(settings, {"theme": "ash"}) + self.assertEqual(cache.get_required_global(settings)["theme"], "ash") + + def test_update_and_delete(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + document = item_doc() + cache = ( + CultCache.builder() + .register_document_type(document) + .add_generic_store(JsonLinesBackingStore(Path(tmp) / "cache.jsonl")) + .build() + ) + cache.pull_all_backing_stores() + cache.put(document, "item:potion", Item(name="Potion", category="Consumable", value=50)) + cache.update(document, "item:potion", lambda item: Item(item.name, item.category, item.value + 5)) + self.assertEqual(cache.get_required(document, "item:potion").value, 55) + cache.delete(document, "item:potion") + self.assertIsNone(cache.get(document, "item:potion")) + + def test_database_entry_formatter_uses_slot_indexed_messagepack_array(self) -> None: + try: + import msgpack # type: ignore + except ModuleNotFoundError: + self.skipTest("msgpack optional dependency is not installed") + + @dataclass + class Settings: + theme: str + retries: int = 0 + + document = define_database_entry_type( + "settings", + [ + ("theme", 0), + ("retries", 2, 0), + ], + cls=Settings, + ) + + payload = document.encode_payload(Settings(theme="ash", retries=3)) + 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_messagepack_store_writes_v1_snapshot(self) -> None: + try: + import msgpack # type: ignore + except ModuleNotFoundError: + self.skipTest("msgpack optional dependency is not installed") + + with tempfile.TemporaryDirectory() as tmp: + document = define_database_entry_type( + "settings", + [("theme", 0)], + schema_id="settings", + schema_name="settings", + schema_version="settings.v1", + ) + store_path = Path(tmp) / "cache.msgpack" + cache = ( + CultCache.builder() + .register_document_type(document) + .add_generic_store(SingleFileMessagePackBackingStore(store_path)) + .build() + ) + cache.pull_all_backing_stores() + cache.put(document, "app", {"theme": "ash"}) + + raw = msgpack.unpackb(store_path.read_bytes(), raw=False) + self.assertEqual(raw[0], "cultcache.store.v1") + self.assertEqual(raw[1][0][0], "settings") + self.assertEqual(raw[2][0][0], "app") + self.assertEqual(raw[2][0][1], "settings") + + def test_interop_cli_helpers_round_trip_v1_store(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + store_path = str(Path(tmp) / "cache.msgpack") + written = write_note(store_path, "python-test") + loaded = read_note(store_path) + self.assertEqual(loaded["documentId"], written["documentId"]) + self.assertEqual(loaded["authorRuntimeId"], "python-test") + self.assertIn("interop", loaded["tags"]) + + def test_cultnet_schema_message_frame_round_trip(self) -> None: + message = hello(runtime_id="python-test", supported_schema_versions=["cultnet.hello.v0"]) + parsed = parse_message(decode_frame(encode_frame(message.to_bytes()))) + self.assertEqual(parsed.schema_version, "cultnet.hello.v0") + self.assertEqual(parsed.body["runtimeId"], "python-test") + + def test_cultmesh_node_uses_cultcache_store(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + document = define_database_entry_type("mesh.note", [("body", 0)]) + store_path = Path(tmp) / "mesh.cc" + node = create_node(store_path, runtime_id="mesh-test") + node.register_document(document) + node.put(document, "note:1", {"body": "hello"}) + + reopened = create_node(store_path, runtime_id="mesh-test") + reopened.register_document(document) + reopened.pull() + self.assertEqual(reopened.get(document, "note:1")["body"], "hello") + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/cultcache-ts/test/cult-cache.test.ts b/packages/cultcache-ts/test/cult-cache.test.ts index acf87ab5..1067eecd 100644 --- a/packages/cultcache-ts/test/cult-cache.test.ts +++ b/packages/cultcache-ts/test/cult-cache.test.ts @@ -20,8 +20,11 @@ const execFileAsync = promisify(execFile); const execAsync = promisify(exec); const cargoCommand = process.env.CARGO ?? (process.platform === "win32" ? join(homedir(), ".cargo", "bin", "cargo.exe") : "cargo"); const dotnetCommand = process.env.DOTNET ?? (process.platform === "win32" ? join("C:", "Program Files", "dotnet", "dotnet.exe") : "dotnet"); +const pythonCommand = process.env.PYTHON ?? "python"; const cultCacheTsRoot = resolve(__dirname, "../.."); const cultLibRoot = findAncestor(cultCacheTsRoot, "CultLib.sln") ?? resolve(cultCacheTsRoot, "..", "CultLib"); +const cultcachePyRoot = resolve(cultLibRoot, "packages", "cultcache-py"); +const cultcachePySrc = resolve(cultcachePyRoot, "src"); const cultcacheRsRoot = existsSync(resolve(cultLibRoot, "packages", "cultcache-rs")) ? resolve(cultLibRoot, "packages", "cultcache-rs") : resolve(cultCacheTsRoot, "..", "cultcache-rs"); @@ -476,7 +479,7 @@ test("SingleFileMessagePackBackingStore heals legacy envelopes whose payload was assert.ok(records[0]?.[3] instanceof Uint8Array); }); -test("CultCache v1 MessagePack stores are readable across TS, Rust, and C#", async () => { +test("CultCache v1 MessagePack stores are readable across TS, Rust, C#, and Python", async () => { await buildInteropPeers(); const tempDir = await mkdtemp(join(tmpdir(), "cultcache-interop-")); const writers = [ @@ -501,6 +504,15 @@ test("CultCache v1 MessagePack stores are readable across TS, Rust, and C#", asy "--runtime-id", "csharp-writer", ], cultLibRoot), }, + { + name: "python", + write: async (file: string) => runJsonCommand("python-write", pythonCommand, [ + "-m", "cultcache_py.interop", + "write", + "--file", file, + "--runtime-id", "python-writer", + ], cultcachePyRoot, { PYTHONPATH: cultcachePySrc }), + }, ]; const readers = [ { @@ -522,6 +534,14 @@ test("CultCache v1 MessagePack stores are readable across TS, Rust, and C#", asy "--file", file, ], cultLibRoot), }, + { + name: "python", + read: async (file: string) => runJsonCommand("python-read", pythonCommand, [ + "-m", "cultcache_py.interop", + "read", + "--file", file, + ], cultcachePyRoot, { PYTHONPATH: cultcachePySrc }), + }, ]; for (const writer of writers) { @@ -732,8 +752,13 @@ async function runJsonCommand( command: string, args: string[], cwd: string, + env: NodeJS.ProcessEnv = {}, ): Promise { - const { stdout, stderr } = await execFileAsync(command, args, { cwd, timeout: 30_000 }); + const { stdout, stderr } = await execFileAsync(command, args, { + cwd, + env: { ...process.env, ...env }, + timeout: 30_000, + }); const trimmed = stdout.trim(); if (!trimmed) { throw new Error(`${name} produced no stdout.\n${stderr}`);