refactor: move serialization adapters into store implementations

Move store-specific serialization adapters from shared serialization.py
into their respective store implementations:
- FullJsonAdapter → Redis store
- StringifiedDictAdapter → MongoDB store
- ElasticsearchAdapter → Elasticsearch store

This improves code organization by keeping serialization logic with
the stores that use them, reducing coupling and making the codebase
more maintainable.

🤖 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

@@ -1,19 +1,19 @@
 import logging
+from abc import ABC, abstractmethod
 from collections.abc import Sequence
 from datetime import datetime
 from typing import Any, overload
 
 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
+from key_value.shared.utils.managed_entry import ManagedEntry, load_from_json, verify_dict
 from key_value.shared.utils.sanitize import (
     ALPHANUMERIC_CHARACTERS,
     LOWERCASE_ALPHABET,
     NUMBERS,
     sanitize_string,
 )
-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
 
@@ -85,6 +85,123 @@
 ALLOWED_INDEX_CHARACTERS: str = LOWERCASE_ALPHABET + NUMBERS + "_" + "-" + "."
 
 
+class SerializationAdapter(ABC):
+    """Base class for ManagedEntry serialization adapters.
+
+    Adapters convert ManagedEntry objects to/from store-specific formats,
+    encapsulating the serialization logic and keeping it separate from the
+    ManagedEntry class and store implementations.
+    """
+
+    @abstractmethod
+    def to_storage(self, key: str, entry: ManagedEntry, collection: str | None = None) -> dict[str, Any] | str:
+        """Convert ManagedEntry to store-specific format.
+
+        Args:
+            key: The key associated with this entry.
+            entry: The ManagedEntry to serialize.
+            collection: Optional collection name for stores that include it in documents.
+
+        Returns:
+            Store-specific format (dict or string depending on store requirements).
+        """
+
+    @abstractmethod
+    def from_storage(self, data: dict[str, Any] | str) -> ManagedEntry:
+        """Convert store-specific format back to ManagedEntry.
+
+        Args:
+            data: The store-specific data to deserialize.
+
+        Returns:
+            A reconstructed ManagedEntry object.
+
+        Raises:
+            DeserializationError: If the data cannot be deserialized.
+        """
+
+
+class ElasticsearchAdapter(SerializationAdapter):
+    """Adapter for Elasticsearch with support for both native and stringified storage.
+
+    Elasticsearch supports two storage modes:
+    - Native (flattened field type): Stores value as native dict for querying
+    - Stringified (disabled object field): Stores value as JSON string for compatibility
+
+    The adapter can read from both fields for backward compatibility but writes to
+    only one based on the native_storage parameter.
+    """
+
+    def __init__(self, native_storage: bool = True) -> None:
+        """Initialize the Elasticsearch adapter.
+
+        Args:
+            native_storage: If True, use flattened field for native dict storage.
+                           If False, use string field for JSON string storage.
+        """
+        self.native_storage = native_storage
+
+    def to_storage(self, key: str, entry: ManagedEntry, collection: str | None = None) -> dict[str, Any]:
+        """Serialize to Elasticsearch document format."""
+        document: dict[str, Any] = {"collection": collection or "", "key": key, "value": {}}
+
+        # Store in appropriate field based on mode
+        if self.native_storage:
+            document["value"]["flattened"] = entry.value_as_dict
+        else:
+            document["value"]["string"] = entry.value_as_json
+
+        if entry.created_at:
+            document["created_at"] = entry.created_at.isoformat()
+        if entry.expires_at:
+            document["expires_at"] = entry.expires_at.isoformat()
+
+        return document
+
+    def from_storage(self, data: dict[str, Any] | str) -> ManagedEntry:
+        """Deserialize from Elasticsearch document format.
+
+        Supports reading from both flattened and string fields for backward compatibility.
+        """
+        if not isinstance(data, dict):
+            msg = "Expected dict data for ElasticsearchAdapter"
+            raise DeserializationError(msg)
+
+        value: dict[str, Any] = {}
+        raw_value = data.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)
+
+        value_flattened = raw_value.get("flattened")
+        if value_flattened:
+            value = verify_dict(obj=value_flattened)
+        else:
+            value_str = raw_value.get("string")
+            if value_str:
+                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 = "Neither flattened nor string field found in value"
+                raise DeserializationError(msg)
+
+        # Import here to avoid circular dependency
+        from key_value.shared.utils.time_to_live import try_parse_datetime_str
+
+        created_at = try_parse_datetime_str(value=data.get("created_at"))
+        expires_at = try_parse_datetime_str(value=data.get("expires_at"))
+
+        return ManagedEntry(
+            value=value,
+            created_at=created_at,
+            expires_at=expires_at,
+        )
+
+
 class ElasticsearchStore(
     BaseEnumerateCollectionsStore, BaseEnumerateKeysStore, BaseDestroyCollectionStore, BaseCullStore, BaseContextManagerStore, BaseStore
 ):

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

@@ -1,10 +1,11 @@
+from abc import ABC, abstractmethod
 from collections.abc import Sequence
 from datetime import datetime
 from typing import Any, overload
 
+from key_value.shared.errors import DeserializationError
 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
@@ -34,6 +35,64 @@
 COLLECTION_ALLOWED_CHARACTERS = ALPHANUMERIC_CHARACTERS + "_"
 
 
+class SerializationAdapter(ABC):
+    """Base class for ManagedEntry serialization adapters.
+
+    Adapters convert ManagedEntry objects to/from store-specific formats,
+    encapsulating the serialization logic and keeping it separate from the
+    ManagedEntry class and store implementations.
+    """
+
+    @abstractmethod
+    def to_storage(self, key: str, entry: ManagedEntry, collection: str | None = None) -> dict[str, Any] | str:
+        """Convert ManagedEntry to store-specific format.
+
+        Args:
+            key: The key associated with this entry.
+            entry: The ManagedEntry to serialize.
+            collection: Optional collection name for stores that include it in documents.
+
+        Returns:
+            Store-specific format (dict or string depending on store requirements).
+        """
+
+    @abstractmethod
+    def from_storage(self, data: dict[str, Any] | str) -> ManagedEntry:
+        """Convert store-specific format back to ManagedEntry.
+
+        Args:
+            data: The store-specific data to deserialize.
+
+        Returns:
+            A reconstructed ManagedEntry object.
+
+        Raises:
+            DeserializationError: If the data cannot be deserialized.
+        """
+
+
+class StringifiedDictAdapter(SerializationAdapter):
+    """Adapter that stores metadata as dict fields but stringifies the value field.
+
+    Used by MongoDB and other document databases that prefer stringified values
+    for consistency or to avoid schema issues with dynamic value structures.
+    """
+
+    def to_storage(self, key: str, entry: ManagedEntry, collection: str | None = None) -> dict[str, Any]:  # noqa: ARG002
+        """Serialize to dict with stringified value field."""
+        return {
+            "key": key,
+            **entry.to_dict(include_metadata=True, include_expiration=True, include_creation=True, stringify_value=True),
+        }
+
+    def from_storage(self, data: dict[str, Any] | str) -> ManagedEntry:
+        """Deserialize from dict with stringified value field."""
+        if not isinstance(data, dict):
+            msg = "Expected dict data for StringifiedDictAdapter"
+            raise DeserializationError(msg)
+        return ManagedEntry.from_dict(data=data, stringified_value=True)
+
+
 class MongoDBStore(BaseEnumerateCollectionsStore, BaseDestroyCollectionStore, BaseContextManagerStore, BaseStore):
     """MongoDB-based key-value store using Motor (async MongoDB driver)."""
 

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

@@ -1,12 +1,13 @@
+from abc import ABC, abstractmethod
 from collections.abc import Sequence
 from datetime import datetime
 from typing import Any, overload
 from urllib.parse import urlparse
 
+from key_value.shared.errors import DeserializationError
 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
@@ -21,6 +22,61 @@
 PAGE_LIMIT = 10000
 
 
+class SerializationAdapter(ABC):
+    """Base class for ManagedEntry serialization adapters.
+
+    Adapters convert ManagedEntry objects to/from store-specific formats,
+    encapsulating the serialization logic and keeping it separate from the
+    ManagedEntry class and store implementations.
+    """
+
+    @abstractmethod
+    def to_storage(self, key: str, entry: ManagedEntry, collection: str | None = None) -> dict[str, Any] | str:
+        """Convert ManagedEntry to store-specific format.
+
+        Args:
+            key: The key associated with this entry.
+            entry: The ManagedEntry to serialize.
+            collection: Optional collection name for stores that include it in documents.
+
+        Returns:
+            Store-specific format (dict or string depending on store requirements).
+        """
+
+    @abstractmethod
+    def from_storage(self, data: dict[str, Any] | str) -> ManagedEntry:
+        """Convert store-specific format back to ManagedEntry.
+
+        Args:
+            data: The store-specific data to deserialize.
+
+        Returns:
+            A reconstructed ManagedEntry object.
+
+        Raises:
+            DeserializationError: If the data cannot be deserialized.
+        """
+
+
+class FullJsonAdapter(SerializationAdapter):
+    """Adapter that serializes to full JSON string with all metadata.
+
+    Used by stores that cannot store native dict structures (e.g., Redis, Valkey, Memcached).
+    The entire entry is serialized to a JSON string including value, TTL, and timestamps.
+    """
+
+    def to_storage(self, key: str, entry: ManagedEntry, collection: str | None = None) -> str:  # noqa: ARG002
+        """Serialize to JSON string with full metadata."""
+        return entry.to_json(include_metadata=True, include_expiration=True, include_creation=True)
+
+    def from_storage(self, data: dict[str, Any] | str) -> ManagedEntry:
+        """Deserialize from JSON string."""
+        if not isinstance(data, str):
+            msg = "Expected string data for FullJsonAdapter"
+            raise DeserializationError(msg)
+        return ManagedEntry.from_json(json_str=data, includes_metadata=True)
+
+
 class RedisStore(BaseDestroyStore, BaseEnumerateKeysStore, BaseContextManagerStore, BaseStore):
     """Redis-based key-value store."""
 

key-value/key-value-aio/tests/stores/elasticsearch/test_elasticsearch.py

@@ -11,7 +11,7 @@
 
 from key_value.aio.stores.base import BaseStore
 from key_value.aio.stores.elasticsearch import ElasticsearchStore
-from key_value.shared.utils.serialization import ElasticsearchAdapter
+from key_value.aio.stores.elasticsearch.store import ElasticsearchAdapter
 from tests.conftest import docker_container, should_skip_docker_tests
 from tests.stores.base import BaseStoreTests, ContextManagerStoreTestMixin
 

key-value/key-value-aio/tests/stores/mongodb/test_mongodb.py

@@ -13,7 +13,7 @@
 
 from key_value.aio.stores.base import BaseStore
 from key_value.aio.stores.mongodb import MongoDBStore
-from key_value.shared.utils.serialization import StringifiedDictAdapter
+from key_value.aio.stores.mongodb.store import StringifiedDictAdapter
 from tests.conftest import docker_container, should_skip_docker_tests
 from tests.stores.base import BaseStoreTests, ContextManagerStoreTestMixin
 

key-value/key-value-aio/tests/stores/redis/test_redis.py

@@ -11,7 +11,7 @@
 
 from key_value.aio.stores.base import BaseStore
 from key_value.aio.stores.redis import RedisStore
-from key_value.shared.utils.serialization import FullJsonAdapter
+from key_value.aio.stores.redis.store import FullJsonAdapter
 from tests.conftest import docker_container, should_skip_docker_tests
 from tests.stores.base import BaseStoreTests, ContextManagerStoreTestMixin