Refactor ManagedEntry serialization with adapter pattern

This commit implements a comprehensive refactoring of ManagedEntry
serialization to address inconsistencies across stores and wrappers.

Changes:
- Created SerializationAdapter base class with three implementations:
  - FullJsonAdapter: For Redis/Valkey (full JSON string)
  - StringifiedDictAdapter: For MongoDB (dict with stringified value)
  - ElasticsearchAdapter: For Elasticsearch (native or stringified)

- Updated stores to use serialization adapters:
  - MongoDB: Uses StringifiedDictAdapter
  - Elasticsearch: Uses ElasticsearchAdapter with native_storage flag
  - Redis: Uses FullJsonAdapter

- Fixed wrapper size calculation inefficiency:
  - Added estimate_serialized_size() utility function
  - Updated CompressionWrapper and LimitSizeWrapper to use utility
  - Eliminates unnecessary ManagedEntry instantiation

- Removed deprecated helper functions:
  - Removed managed_entry_to_document/document_to_managed_entry from MongoDB
  - Removed managed_entry_to_document/source_to_managed_entry from Elasticsearch
  - Removed managed_entry_to_json/json_to_managed_entry from Redis

- Updated test files to use new adapter-based approach

Benefits:
- Consistent serialization interface across all stores
- Separation of concerns (serialization logic in adapters)
- Easier to add new stores or modify serialization
- More efficient size calculations in wrappers
- Better testability and maintainability

Related to issue #165

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: William Easton <[email protected]>

Author: github-actions[bot]

key-value/key-value-aio/src/key_value/aio/stores/elasticsearch/store.py

@@ -6,14 +6,15 @@
 from elastic_transport import ObjectApiResponse
 from elastic_transport import SerializationError as ElasticsearchSerializationError
 from key_value.shared.errors import DeserializationError, SerializationError
-from key_value.shared.utils.managed_entry import ManagedEntry, load_from_json, verify_dict
+from key_value.shared.utils.managed_entry import ManagedEntry
 from key_value.shared.utils.sanitize import (
     ALPHANUMERIC_CHARACTERS,
     LOWERCASE_ALPHABET,
     NUMBERS,
     sanitize_string,
 )
-from key_value.shared.utils.time_to_live import now_as_epoch, try_parse_datetime_str
+from key_value.shared.utils.serialization import ElasticsearchAdapter, SerializationAdapter
+from key_value.shared.utils.time_to_live import now_as_epoch
 from typing_extensions import override
 
 from key_value.aio.stores.base import (
@@ -84,54 +85,6 @@
 ALLOWED_INDEX_CHARACTERS: str = LOWERCASE_ALPHABET + NUMBERS + "_" + "-" + "."
 
 
-def managed_entry_to_document(collection: str, key: str, managed_entry: ManagedEntry, *, native_storage: bool = False) -> dict[str, Any]:
-    document: dict[str, Any] = {"collection": collection, "key": key, "value": {}}
-
-    # Store in appropriate field based on mode
-    if native_storage:
-        document["value"]["flattened"] = managed_entry.value_as_dict
-    else:
-        document["value"]["string"] = managed_entry.value_as_json
-
-    if managed_entry.created_at:
-        document["created_at"] = managed_entry.created_at.isoformat()
-    if managed_entry.expires_at:
-        document["expires_at"] = managed_entry.expires_at.isoformat()
-
-    return document
-
-
-def source_to_managed_entry(source: dict[str, Any]) -> ManagedEntry:
-    value: dict[str, Any] = {}
-
-    raw_value = source.get("value")
-
-    # Try flattened field first, fall back to string field
-    if not raw_value or not isinstance(raw_value, dict):
-        msg = "Value field not found or invalid type"
-        raise DeserializationError(msg)
-
-    if value_flattened := raw_value.get("flattened"):  # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType]
-        value = verify_dict(obj=value_flattened)
-    elif value_str := raw_value.get("string"):  # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType]
-        if not isinstance(value_str, str):
-            msg = "Value in `value` field is not a string"
-            raise DeserializationError(msg)
-        value = load_from_json(value_str)
-    else:
-        msg = "Value field not found or invalid type"
-        raise DeserializationError(msg)
-
-    created_at: datetime | None = try_parse_datetime_str(value=source.get("created_at"))
-    expires_at: datetime | None = try_parse_datetime_str(value=source.get("expires_at"))
-
-    return ManagedEntry(
-        value=value,
-        created_at=created_at,
-        expires_at=expires_at,
-    )
-
-
 class ElasticsearchStore(
     BaseEnumerateCollectionsStore, BaseEnumerateKeysStore, BaseDestroyCollectionStore, BaseCullStore, BaseContextManagerStore, BaseStore
 ):
@@ -145,6 +98,8 @@ class ElasticsearchStore(
 
     _native_storage: bool
 
+    _adapter: SerializationAdapter
+
     @overload
     def __init__(
         self,
@@ -208,6 +163,7 @@ def __init__(
         self._index_prefix = index_prefix
         self._native_storage = native_storage
         self._is_serverless = False
+        self._adapter = ElasticsearchAdapter(native_storage=native_storage)
 
         super().__init__(default_collection=default_collection)
 
@@ -260,7 +216,7 @@ async def _get_managed_entry(self, *, key: str, collection: str) -> ManagedEntry
             return None
 
         try:
-            return source_to_managed_entry(source=source)
+            return self._adapter.from_storage(data=source)
         except DeserializationError:
             return None
 
@@ -293,7 +249,7 @@ async def _get_managed_entries(self, *, collection: str, keys: Sequence[str]) ->
                 continue
 
             try:
-                entries_by_id[doc_id] = source_to_managed_entry(source=source)
+                entries_by_id[doc_id] = self._adapter.from_storage(data=source)
             except DeserializationError as e:
                 logger.error(
                     "Failed to deserialize Elasticsearch document in batch operation",
@@ -324,9 +280,11 @@ async def _put_managed_entry(
         index_name: str = self._sanitize_index_name(collection=collection)
         document_id: str = self._sanitize_document_id(key=key)
 
-        document: dict[str, Any] = managed_entry_to_document(
-            collection=collection, key=key, managed_entry=managed_entry, native_storage=self._native_storage
-        )
+        document = self._adapter.to_storage(collection=collection, key=key, entry=managed_entry)
+
+        if not isinstance(document, dict):
+            msg = "Elasticsearch adapter must return dict"
+            raise TypeError(msg)
 
         try:
             _ = await self._client.index(
@@ -364,9 +322,11 @@ async def _put_managed_entries(
 
             index_action: dict[str, Any] = new_bulk_action(action="index", index=index_name, document_id=document_id)
 
-            document: dict[str, Any] = managed_entry_to_document(
-                collection=collection, key=key, managed_entry=managed_entry, native_storage=self._native_storage
-            )
+            document = self._adapter.to_storage(collection=collection, key=key, entry=managed_entry)
+
+            if not isinstance(document, dict):
+                msg = "Elasticsearch adapter must return dict"
+                raise TypeError(msg)
 
             operations.extend([index_action, document])
         try:

key-value/key-value-aio/src/key_value/aio/stores/mongodb/store.py

@@ -4,6 +4,7 @@
 
 from key_value.shared.utils.managed_entry import ManagedEntry
 from key_value.shared.utils.sanitize import ALPHANUMERIC_CHARACTERS, sanitize_string
+from key_value.shared.utils.serialization import SerializationAdapter, StringifiedDictAdapter
 from typing_extensions import Self, override
 
 from key_value.aio.stores.base import BaseContextManagerStore, BaseDestroyCollectionStore, BaseEnumerateCollectionsStore, BaseStore
@@ -33,48 +34,13 @@
 COLLECTION_ALLOWED_CHARACTERS = ALPHANUMERIC_CHARACTERS + "_"
 
 
-def document_to_managed_entry(document: dict[str, Any]) -> ManagedEntry:
-    """Convert a MongoDB document back to a ManagedEntry.
-
-    This function deserializes a MongoDB document (created by `managed_entry_to_document`) back to a
-    ManagedEntry object, parsing the stringified value field and preserving all metadata.
-
-    Args:
-        document: The MongoDB document to convert.
-
-    Returns:
-        A ManagedEntry object reconstructed from the document.
-    """
-    return ManagedEntry.from_dict(data=document, stringified_value=True)
-
-
-def managed_entry_to_document(key: str, managed_entry: ManagedEntry) -> dict[str, Any]:
-    """Convert a ManagedEntry to a MongoDB document for storage.
-
-    This function serializes a ManagedEntry to a MongoDB document format, including the key and all
-    metadata (TTL, creation, and expiration timestamps). The value is stringified to ensure proper
-    storage in MongoDB. The serialization is designed to preserve all entry information for round-trip
-    conversion back to a ManagedEntry.
-
-    Args:
-        key: The key associated with this entry.
-        managed_entry: The ManagedEntry to serialize.
-
-    Returns:
-        A MongoDB document dict containing the key, value, and all metadata.
-    """
-    return {
-        "key": key,
-        **managed_entry.to_dict(include_metadata=True, include_expiration=True, include_creation=True, stringify_value=True),
-    }
-
-
 class MongoDBStore(BaseEnumerateCollectionsStore, BaseDestroyCollectionStore, BaseContextManagerStore, BaseStore):
     """MongoDB-based key-value store using Motor (async MongoDB driver)."""
 
     _client: AsyncMongoClient[dict[str, Any]]
     _db: AsyncDatabase[dict[str, Any]]
     _collections_by_name: dict[str, AsyncCollection[dict[str, Any]]]
+    _adapter: SerializationAdapter
 
     @overload
     def __init__(
@@ -131,6 +97,7 @@ def __init__(
 
         self._db = self._client[db_name]
         self._collections_by_name = {}
+        self._adapter = StringifiedDictAdapter()
 
         super().__init__(default_collection=default_collection)
 
@@ -187,7 +154,7 @@ async def _get_managed_entry(self, *, key: str, collection: str) -> ManagedEntry
         sanitized_collection = self._sanitize_collection_name(collection=collection)
 
         if doc := await self._collections_by_name[sanitized_collection].find_one(filter={"key": key}):
-            return ManagedEntry.from_dict(data=doc, stringified_value=True)
+            return self._adapter.from_storage(data=doc)
 
         return None
 
@@ -205,7 +172,7 @@ async def _get_managed_entries(self, *, collection: str, keys: Sequence[str]) ->
 
         async for doc in cursor:
             if key := doc.get("key"):
-                managed_entries_by_key[key] = document_to_managed_entry(document=doc)
+                managed_entries_by_key[key] = self._adapter.from_storage(data=doc)
 
         return [managed_entries_by_key[key] for key in keys]
 
@@ -217,7 +184,11 @@ async def _put_managed_entry(
         collection: str,
         managed_entry: ManagedEntry,
     ) -> None:
-        mongo_doc: dict[str, Any] = managed_entry_to_document(key=key, managed_entry=managed_entry)
+        mongo_doc = self._adapter.to_storage(key=key, entry=managed_entry)
+
+        if not isinstance(mongo_doc, dict):
+            msg = "MongoDB adapter must return dict"
+            raise TypeError(msg)
 
         sanitized_collection = self._sanitize_collection_name(collection=collection)
 
@@ -248,7 +219,11 @@ async def _put_managed_entries(
 
         operations: list[UpdateOne] = []
         for key, managed_entry in zip(keys, managed_entries, strict=True):
-            mongo_doc: dict[str, Any] = managed_entry_to_document(key=key, managed_entry=managed_entry)
+            mongo_doc = self._adapter.to_storage(key=key, entry=managed_entry)
+
+            if not isinstance(mongo_doc, dict):
+                msg = "MongoDB adapter must return dict"
+                raise TypeError(msg)
 
             operations.append(
                 UpdateOne(

key-value/key-value-aio/src/key_value/aio/stores/redis/store.py

@@ -6,6 +6,7 @@
 from key_value.shared.type_checking.bear_spray import bear_spray
 from key_value.shared.utils.compound import compound_key, get_keys_from_compound_keys
 from key_value.shared.utils.managed_entry import ManagedEntry
+from key_value.shared.utils.serialization import FullJsonAdapter, SerializationAdapter
 from typing_extensions import override
 
 from key_value.aio.stores.base import BaseContextManagerStore, BaseDestroyStore, BaseEnumerateKeysStore, BaseStore
@@ -20,41 +21,11 @@
 PAGE_LIMIT = 10000
 
 
-def managed_entry_to_json(managed_entry: ManagedEntry) -> str:
-    """Convert a ManagedEntry to a JSON string for Redis storage.
-
-    This function serializes a ManagedEntry to JSON format including all metadata (TTL, creation,
-    and expiration timestamps). The serialization is designed to preserve all entry information
-    for round-trip conversion back to a ManagedEntry.
-
-    Args:
-        managed_entry: The ManagedEntry to serialize.
-
-    Returns:
-        A JSON string representation of the ManagedEntry with full metadata.
-    """
-    return managed_entry.to_json(include_metadata=True, include_expiration=True, include_creation=True)
-
-
-def json_to_managed_entry(json_str: str) -> ManagedEntry:
-    """Convert a JSON string from Redis storage back to a ManagedEntry.
-
-    This function deserializes a JSON string (created by `managed_entry_to_json`) back to a
-    ManagedEntry object, preserving all metadata including TTL, creation, and expiration timestamps.
-
-    Args:
-        json_str: The JSON string to deserialize.
-
-    Returns:
-        A ManagedEntry object reconstructed from the JSON string.
-    """
-    return ManagedEntry.from_json(json_str=json_str, includes_metadata=True)
-
-
 class RedisStore(BaseDestroyStore, BaseEnumerateKeysStore, BaseContextManagerStore, BaseStore):
     """Redis-based key-value store."""
 
     _client: Redis
+    _adapter: SerializationAdapter
 
     @overload
     def __init__(self, *, client: Redis, default_collection: str | None = None) -> None: ...
@@ -111,6 +82,7 @@ def __init__(
             )
 
         self._stable_api = True
+        self._adapter = FullJsonAdapter()
 
         super().__init__(default_collection=default_collection)
 
@@ -123,7 +95,7 @@ async def _get_managed_entry(self, *, key: str, collection: str) -> ManagedEntry
         if not isinstance(redis_response, str):
             return None
 
-        managed_entry: ManagedEntry = json_to_managed_entry(json_str=redis_response)
+        managed_entry: ManagedEntry = self._adapter.from_storage(data=redis_response)
 
         return managed_entry
 
@@ -139,7 +111,7 @@ async def _get_managed_entries(self, *, collection: str, keys: Sequence[str]) ->
         entries: list[ManagedEntry | None] = []
         for redis_response in redis_responses:
             if isinstance(redis_response, str):
-                entries.append(json_to_managed_entry(json_str=redis_response))
+                entries.append(self._adapter.from_storage(data=redis_response))
             else:
                 entries.append(None)
 
@@ -155,7 +127,11 @@ async def _put_managed_entry(
     ) -> None:
         combo_key: str = compound_key(collection=collection, key=key)
 
-        json_value: str = managed_entry_to_json(managed_entry=managed_entry)
+        json_value = self._adapter.to_storage(key=key, entry=managed_entry)
+
+        if not isinstance(json_value, str):
+            msg = "Redis adapter must return str"
+            raise TypeError(msg)
 
         if managed_entry.ttl is not None:
             # Redis does not support <= 0 TTLs
@@ -181,10 +157,13 @@ async def _put_managed_entries(
 
         if ttl is None:
             # If there is no TTL, we can just do a simple mset
-            mapping: dict[str, str] = {
-                compound_key(collection=collection, key=key): managed_entry_to_json(managed_entry=managed_entry)
-                for key, managed_entry in zip(keys, managed_entries, strict=True)
-            }
+            mapping: dict[str, str] = {}
+            for key, managed_entry in zip(keys, managed_entries, strict=True):
+                json_value = self._adapter.to_storage(key=key, entry=managed_entry)
+                if not isinstance(json_value, str):
+                    msg = "Redis adapter must return str"
+                    raise TypeError(msg)
+                mapping[compound_key(collection=collection, key=key)] = json_value
 
             await self._client.mset(mapping=mapping)
 
@@ -198,7 +177,11 @@ async def _put_managed_entries(
 
         for key, managed_entry in zip(keys, managed_entries, strict=True):
             combo_key: str = compound_key(collection=collection, key=key)
-            json_value: str = managed_entry_to_json(managed_entry=managed_entry)
+            json_value = self._adapter.to_storage(key=key, entry=managed_entry)
+
+            if not isinstance(json_value, str):
+                msg = "Redis adapter must return str"
+                raise TypeError(msg)
 
             pipeline.setex(name=combo_key, time=ttl_seconds, value=json_value)
 

key-value/key-value-aio/src/key_value/aio/wrappers/compression/wrapper.py

@@ -4,7 +4,7 @@
 from collections.abc import Mapping, Sequence
 from typing import Any, SupportsFloat
 
-from key_value.shared.utils.managed_entry import ManagedEntry
+from key_value.shared.utils.managed_entry import estimate_serialized_size
 from typing_extensions import override
 
 from key_value.aio.protocols.key_value import AsyncKeyValue
@@ -56,7 +56,7 @@ def _should_compress(self, value: dict[str, Any]) -> bool:
             return False
 
         # Check size
-        item_size: int = len(ManagedEntry(value=value).to_json())
+        item_size: int = estimate_serialized_size(value=value)
         return item_size >= self.min_size_to_compress
 
     def _compress_value(self, value: dict[str, Any]) -> dict[str, Any]: