refactor: align S3 store with sanitization strategy pattern

- Add S3KeySanitizationStrategy and S3CollectionSanitizationStrategy classes
- Remove automatic sanitization from _get_s3_key()
- Add collection_sanitization_strategy and key_sanitization_strategy parameters to S3Store
- Update S3Store to use BaseStore's _sanitize_collection_and_key() method
- By default, collections/keys are not sanitized (matches new main pattern)
- Update tests to use sanitization strategies
- Update documentation to explain when/why to use strategies

This aligns with the main branch refactor where stores no longer
sanitize collections and keys by default, making sanitization opt-in
via strategy parameters.

Co-authored-by: William Easton <strawgate@users.noreply.github.com>

Author: github-actions[bot]

key-value/key-value-aio/src/key_value/aio/stores/s3/__init__.py

@@ -1,5 +1,13 @@
 """AWS S3-based key-value store."""
 
-from key_value.aio.stores.s3.store import S3Store
+from key_value.aio.stores.s3.store import (
+    S3CollectionSanitizationStrategy,
+    S3KeySanitizationStrategy,
+    S3Store,
+)
 
-__all__ = ["S3Store"]
+__all__ = [
+    "S3CollectionSanitizationStrategy",
+    "S3KeySanitizationStrategy",
+    "S3Store",
+]

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

@@ -2,6 +2,7 @@
 from typing import TYPE_CHECKING, Any, overload
 
 from key_value.shared.utils.managed_entry import ManagedEntry
+from key_value.shared.utils.sanitization import SanitizationStrategy
 from key_value.shared.utils.sanitize import hash_excess_length
 from typing_extensions import Self, override
 
@@ -33,6 +34,55 @@
     from aiobotocore.client import AioBaseClient as S3Client
 
 
+class S3KeySanitizationStrategy(SanitizationStrategy):
+    """Sanitization strategy for S3 keys with byte-aware length limits.
+
+    S3 has a maximum key length of 1024 bytes (UTF-8 encoded). This strategy
+    hashes keys that exceed the specified byte limit to ensure compliance.
+
+    Args:
+        max_bytes: Maximum key length in bytes. Defaults to 500.
+    """
+
+    def __init__(self, max_bytes: int = MAX_KEY_LENGTH) -> None:
+        """Initialize the S3 key sanitization strategy.
+
+        Args:
+            max_bytes: Maximum key length in bytes.
+        """
+        self.max_bytes = max_bytes
+
+    def sanitize(self, value: str) -> str:
+        """Hash the value if it exceeds max_bytes when UTF-8 encoded.
+
+        Args:
+            value: The key to sanitize.
+
+        Returns:
+            The original value if within limit, or truncated+hashed if too long.
+        """
+        return hash_excess_length(value, self.max_bytes, length_is_bytes=True)
+
+    def validate(self, value: str) -> None:
+        """No validation needed for S3 keys."""
+
+
+class S3CollectionSanitizationStrategy(S3KeySanitizationStrategy):
+    """Sanitization strategy for S3 collection names with byte-aware length limits.
+
+    This is identical to S3KeySanitizationStrategy but uses a default of 500 bytes
+    for collection names to match the S3 key format {collection}/{key}.
+    """
+
+    def __init__(self, max_bytes: int = MAX_COLLECTION_LENGTH) -> None:
+        """Initialize the S3 collection sanitization strategy.
+
+        Args:
+            max_bytes: Maximum collection name length in bytes.
+        """
+        super().__init__(max_bytes=max_bytes)
+
+
 class S3Store(BaseContextManagerStore, BaseStore):
     """AWS S3-based key-value store.
 
@@ -42,13 +92,28 @@ class S3Store(BaseContextManagerStore, BaseStore):
     S3 object metadata and checked client-side during retrieval (S3 lifecycle policies
     can be configured separately for background cleanup, but don't provide atomic TTL+retrieval).
 
+    By default, collections and keys are not sanitized. This means you must ensure that
+    the combined "{collection}/{key}" path does not exceed S3's 1024-byte limit when UTF-8 encoded.
+
+    To handle long collection or key names, use the S3CollectionSanitizationStrategy and
+    S3KeySanitizationStrategy which will hash values exceeding the byte limit.
+
     Example:
         Basic usage with automatic AWS credentials:
 
         >>> async with S3Store(bucket_name="my-kv-store") as store:
         ...     await store.put(key="user:123", value={"name": "Alice"}, ttl=3600)
         ...     user = await store.get(key="user:123")
 
+        With sanitization for long keys/collections:
+
+        >>> async with S3Store(
+        ...     bucket_name="my-kv-store",
+        ...     collection_sanitization_strategy=S3CollectionSanitizationStrategy(),
+        ...     key_sanitization_strategy=S3KeySanitizationStrategy(),
+        ... ) as store:
+        ...     await store.put(key="very_long_key" * 100, value={"data": "test"})
+
         With custom AWS credentials:
 
         >>> async with S3Store(
@@ -74,13 +139,23 @@ class S3Store(BaseContextManagerStore, BaseStore):
     _client: S3Client | None
 
     @overload
-    def __init__(self, *, client: S3Client, bucket_name: str, default_collection: str | None = None) -> None:
+    def __init__(
+        self,
+        *,
+        client: S3Client,
+        bucket_name: str,
+        default_collection: str | None = None,
+        collection_sanitization_strategy: SanitizationStrategy | None = None,
+        key_sanitization_strategy: SanitizationStrategy | None = None,
+    ) -> None:
         """Initialize the S3 store with a pre-configured client.
 
         Args:
             client: The S3 client to use. You must have entered the context manager before passing this in.
             bucket_name: The name of the S3 bucket to use.
             default_collection: The default collection to use if no collection is provided.
+            collection_sanitization_strategy: Strategy for sanitizing collection names. Defaults to None (no sanitization).
+            key_sanitization_strategy: Strategy for sanitizing keys. Defaults to None (no sanitization).
         """
 
     @overload
@@ -94,6 +169,8 @@ def __init__(
         aws_secret_access_key: str | None = None,
         aws_session_token: str | None = None,
         default_collection: str | None = None,
+        collection_sanitization_strategy: SanitizationStrategy | None = None,
+        key_sanitization_strategy: SanitizationStrategy | None = None,
     ) -> None:
         """Initialize the S3 store with AWS credentials.
 
@@ -105,6 +182,8 @@ def __init__(
             aws_secret_access_key: AWS secret access key. Defaults to None (uses AWS default credentials).
             aws_session_token: AWS session token. Defaults to None (uses AWS default credentials).
             default_collection: The default collection to use if no collection is provided.
+            collection_sanitization_strategy: Strategy for sanitizing collection names. Defaults to None (no sanitization).
+            key_sanitization_strategy: Strategy for sanitizing keys. Defaults to None (no sanitization).
         """
 
     def __init__(
@@ -118,6 +197,8 @@ def __init__(
         aws_secret_access_key: str | None = None,
         aws_session_token: str | None = None,
         default_collection: str | None = None,
+        collection_sanitization_strategy: SanitizationStrategy | None = None,
+        key_sanitization_strategy: SanitizationStrategy | None = None,
     ) -> None:
         """Initialize the S3 store.
 
@@ -130,6 +211,8 @@ def __init__(
             aws_secret_access_key: AWS secret access key. Defaults to None (uses AWS default credentials).
             aws_session_token: AWS session token. Defaults to None (uses AWS default credentials).
             default_collection: The default collection to use if no collection is provided.
+            collection_sanitization_strategy: Strategy for sanitizing collection names. Defaults to None (no sanitization).
+            key_sanitization_strategy: Strategy for sanitizing keys. Defaults to None (no sanitization).
         """
         self._bucket_name = bucket_name
         self._endpoint_url = endpoint_url
@@ -148,7 +231,11 @@ def __init__(
             self._raw_client = session.client(service_name="s3", endpoint_url=endpoint_url)  # pyright: ignore[reportUnknownMemberType]
             self._client = None
 
-        super().__init__(default_collection=default_collection)
+        super().__init__(
+            default_collection=default_collection,
+            collection_sanitization_strategy=collection_sanitization_strategy,
+            key_sanitization_strategy=key_sanitization_strategy,
+        )
 
     async def _connect(self) -> None:
         if self._client is None and self._raw_client:
@@ -230,8 +317,8 @@ async def _setup(self) -> None:
     def _get_s3_key(self, *, collection: str, key: str) -> str:
         """Generate the S3 object key for a given collection and key.
 
-        S3 has a maximum key length of 1024 bytes. To ensure compliance, we hash
-        long collection or key names to stay within limits while maintaining uniqueness.
+        The collection and key are sanitized using the configured sanitization strategies
+        before being combined into the S3 object key format: {collection}/{key}.
 
         Args:
             collection: The collection name.
@@ -240,11 +327,9 @@ def _get_s3_key(self, *, collection: str, key: str) -> str:
         Returns:
             The S3 object key in format: {collection}/{key}
         """
-        # Hash collection and key if they exceed their max byte lengths
-        # This ensures the combined S3 key stays under 1024 bytes
-        safe_collection = hash_excess_length(collection, MAX_COLLECTION_LENGTH, length_is_bytes=True)
-        safe_key = hash_excess_length(key, MAX_KEY_LENGTH, length_is_bytes=True)
-        return f"{safe_collection}/{safe_key}"
+        # Use the sanitization strategies from BaseStore
+        sanitized_collection, sanitized_key = self._sanitize_collection_and_key(collection=collection, key=key)
+        return f"{sanitized_collection}/{sanitized_key}"
 
     @override
     async def _get_managed_entry(self, *, key: str, collection: str) -> ManagedEntry | None:

key-value/key-value-aio/tests/stores/s3/test_s3.py

@@ -70,12 +70,17 @@ async def setup_s3(self, request: pytest.FixtureRequest) -> AsyncGenerator[None,
     @override
     @pytest.fixture
     async def store(self, setup_s3: None) -> S3Store:
+        from key_value.aio.stores.s3 import S3CollectionSanitizationStrategy, S3KeySanitizationStrategy
+
         store = S3Store(
             bucket_name=S3_TEST_BUCKET,
             endpoint_url=S3_ENDPOINT,
             aws_access_key_id="test",
             aws_secret_access_key="test",
             region_name="us-east-1",
+            # Use sanitization strategies for tests to handle long collection/key names
+            collection_sanitization_strategy=S3CollectionSanitizationStrategy(),
+            key_sanitization_strategy=S3KeySanitizationStrategy(),
         )
 
         # Clean up test bucket if it exists