fix: Address CodeRabbit PR feedback

- Add explicit list[str] type annotations to fix pyright errors
- Fix max-length/prefix enforcement in CharacterSanitizationStrategy
- Add truncation when add_hash is False
- Ensure S_ prefix when sanitization changes input
- Use local variables to narrow types for None checks
- Add noqa for "too many statements" warning

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

Author: github-actions[bot]

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

@@ -183,7 +183,7 @@ async def get(self, key: str, *, collection: str | None = None) -> dict[str, Any
         return result
 
     @override
-    async def get_many(self, keys: Sequence[str], *, collection: str | None = None) -> list[dict[str, Any] | None]:  # noqa: PLR0912
+    async def get_many(self, keys: Sequence[str], *, collection: str | None = None) -> list[dict[str, Any] | None]:  # noqa: PLR0912, PLR0915
         """Get multiple values, falling back to legacy store for missing keys.
 
         Args:
@@ -194,10 +194,10 @@ async def get_many(self, keys: Sequence[str], *, collection: str | None = None)
             List of values (None for missing keys).
         """
         # Separate keys by cached location
-        current_keys = []
-        legacy_keys = []
-        missing_keys = []
-        unknown_keys = []
+        current_keys: list[str] = []
+        legacy_keys: list[str] = []
+        missing_keys: list[str] = []
+        unknown_keys: list[str] = []
 
         for key in keys:
             cached_location = self._cache_get(key, collection)
@@ -211,7 +211,7 @@ async def get_many(self, keys: Sequence[str], *, collection: str | None = None)
                 unknown_keys.append(key)
 
         # Start with all None
-        key_to_value: dict[str, dict[str, Any] | None] = dict.fromkeys(keys, None)
+        key_to_value: dict[str, dict[str, Any] | None] = dict.fromkeys(keys, None)  # type: ignore[arg-type]
 
         # Fetch known current keys
         if current_keys:
@@ -223,12 +223,13 @@ async def get_many(self, keys: Sequence[str], *, collection: str | None = None)
         if legacy_keys:
             legacy_results = await self.legacy_store.get_many(keys=legacy_keys, collection=collection)
             for i, key in enumerate(legacy_keys):
-                key_to_value[key] = legacy_results[i]
+                legacy_value = legacy_results[i]
+                key_to_value[key] = legacy_value
 
                 # Optionally migrate
-                if self.migrate_on_read and legacy_results[i] is not None:
+                if self.migrate_on_read and legacy_value is not None:
                     _, ttl = await self.legacy_store.ttl(key=key, collection=collection)
-                    await self.current_store.put(key=key, value=legacy_results[i], collection=collection, ttl=ttl)
+                    await self.current_store.put(key=key, value=legacy_value, collection=collection, ttl=ttl)
                     if self.delete_after_migration:
                         await self.legacy_store.delete(key=key, collection=collection)
                     self._cache_put(key, collection, "current")
@@ -238,7 +239,7 @@ async def get_many(self, keys: Sequence[str], *, collection: str | None = None)
             current_results = await self.current_store.get_many(keys=unknown_keys, collection=collection)
 
             # Identify which unknown keys were not found in current
-            not_in_current = []
+            not_in_current: list[str] = []
             for i, key in enumerate(unknown_keys):
                 if current_results[i] is not None:
                     key_to_value[key] = current_results[i]
@@ -250,13 +251,14 @@ async def get_many(self, keys: Sequence[str], *, collection: str | None = None)
             if not_in_current:
                 legacy_results = await self.legacy_store.get_many(keys=not_in_current, collection=collection)
                 for i, key in enumerate(not_in_current):
-                    if legacy_results[i] is not None:
-                        key_to_value[key] = legacy_results[i]
+                    legacy_value = legacy_results[i]
+                    if legacy_value is not None:
+                        key_to_value[key] = legacy_value
 
                         # Optionally migrate
                         if self.migrate_on_read:
                             _, ttl = await self.legacy_store.ttl(key=key, collection=collection)
-                            await self.current_store.put(key=key, value=legacy_results[i], collection=collection, ttl=ttl)
+                            await self.current_store.put(key=key, value=legacy_value, collection=collection, ttl=ttl)
                             if self.delete_after_migration:
                                 await self.legacy_store.delete(key=key, collection=collection)
                             self._cache_put(key, collection, "current")
@@ -324,10 +326,10 @@ async def ttl_many(self, keys: Sequence[str], *, collection: str | None = None)
             List of (value, ttl) tuples.
         """
         # Similar logic to get_many but with TTL
-        current_keys = []
-        legacy_keys = []
-        missing_keys = []
-        unknown_keys = []
+        current_keys: list[str] = []
+        legacy_keys: list[str] = []
+        missing_keys: list[str] = []
+        unknown_keys: list[str] = []
 
         for key in keys:
             cached_location = self._cache_get(key, collection)
@@ -340,7 +342,7 @@ async def ttl_many(self, keys: Sequence[str], *, collection: str | None = None)
             else:
                 unknown_keys.append(key)
 
-        key_to_value: dict[str, tuple[dict[str, Any] | None, float | None]] = dict.fromkeys(keys, (None, None))  # type: ignore
+        key_to_value: dict[str, tuple[dict[str, Any] | None, float | None]] = dict.fromkeys(keys, (None, None))
 
         # Fetch from current
         if current_keys:
@@ -365,7 +367,7 @@ async def ttl_many(self, keys: Sequence[str], *, collection: str | None = None)
         if unknown_keys:
             current_results = await self.current_store.ttl_many(keys=unknown_keys, collection=collection)
 
-            not_in_current = []
+            not_in_current: list[str] = []
             for i, key in enumerate(unknown_keys):
                 if current_results[i][0] is not None:
                     key_to_value[key] = current_results[i]

key-value/key-value-shared/src/key_value/shared/utils/sanitization_strategy.py

@@ -229,7 +229,21 @@ def sanitize(self, value: str) -> str:
             # Add prefix, hash fragment, and return
             return f"S_{sanitized}-{hash_fragment}"
 
-        # No hash needed - return sanitized value (or original if unchanged)
+        # No hash needed - but still need to enforce max_length and prefix if changed
+        needs_truncation = len(sanitized) > self.max_length
+        if needs_truncation:
+            sanitized = sanitized[: self.max_length]
+            changed = True
+
+        if changed:
+            # Add S_ prefix since we modified the input
+            prefixed = f"S_{sanitized}"
+            # Ensure the prefixed result fits within max_length
+            if len(prefixed) > self.max_length:
+                sanitized = sanitized[: self.max_length - 2]
+                prefixed = f"S_{sanitized}"
+            return prefixed
+
         return sanitized
 
     def validate(self, value: str) -> None:

key-value/key-value-sync/src/key_value/sync/code_gen/wrappers/sanitization_migration/wrapper.py

@@ -186,7 +186,7 @@ def get(self, key: str, *, collection: str | None = None) -> dict[str, Any] | No
         return result
 
     @override
-    def get_many(self, keys: Sequence[str], *, collection: str | None = None) -> list[dict[str, Any] | None]:  # noqa: PLR0912
+    def get_many(self, keys: Sequence[str], *, collection: str | None = None) -> list[dict[str, Any] | None]:  # noqa: PLR0912, PLR0915
         """Get multiple values, falling back to legacy store for missing keys.
 
         Args:
@@ -197,10 +197,10 @@ def get_many(self, keys: Sequence[str], *, collection: str | None = None) -> lis
             List of values (None for missing keys).
         """
         # Separate keys by cached location
-        current_keys = []
-        legacy_keys = []
-        missing_keys = []
-        unknown_keys = []
+        current_keys: list[str] = []
+        legacy_keys: list[str] = []
+        missing_keys: list[str] = []
+        unknown_keys: list[str] = []
 
         for key in keys:
             cached_location = self._cache_get(key, collection)
@@ -214,7 +214,7 @@ def get_many(self, keys: Sequence[str], *, collection: str | None = None) -> lis
                 unknown_keys.append(key)
 
         # Start with all None
-        key_to_value: dict[str, dict[str, Any] | None] = dict.fromkeys(keys, None)
+        key_to_value: dict[str, dict[str, Any] | None] = dict.fromkeys(keys, None)  # type: ignore[arg-type]
 
         # Fetch known current keys
         if current_keys:
@@ -226,12 +226,13 @@ def get_many(self, keys: Sequence[str], *, collection: str | None = None) -> lis
         if legacy_keys:
             legacy_results = self.legacy_store.get_many(keys=legacy_keys, collection=collection)
             for i, key in enumerate(legacy_keys):
-                key_to_value[key] = legacy_results[i]
+                legacy_value = legacy_results[i]
+                key_to_value[key] = legacy_value
 
                 # Optionally migrate
-                if self.migrate_on_read and legacy_results[i] is not None:
+                if self.migrate_on_read and legacy_value is not None:
                     (_, ttl) = self.legacy_store.ttl(key=key, collection=collection)
-                    self.current_store.put(key=key, value=legacy_results[i], collection=collection, ttl=ttl)
+                    self.current_store.put(key=key, value=legacy_value, collection=collection, ttl=ttl)
                     if self.delete_after_migration:
                         self.legacy_store.delete(key=key, collection=collection)
                     self._cache_put(key, collection, "current")
@@ -241,7 +242,7 @@ def get_many(self, keys: Sequence[str], *, collection: str | None = None) -> lis
             current_results = self.current_store.get_many(keys=unknown_keys, collection=collection)
 
             # Identify which unknown keys were not found in current
-            not_in_current = []
+            not_in_current: list[str] = []
             for i, key in enumerate(unknown_keys):
                 if current_results[i] is not None:
                     key_to_value[key] = current_results[i]
@@ -253,13 +254,14 @@ def get_many(self, keys: Sequence[str], *, collection: str | None = None) -> lis
             if not_in_current:
                 legacy_results = self.legacy_store.get_many(keys=not_in_current, collection=collection)
                 for i, key in enumerate(not_in_current):
-                    if legacy_results[i] is not None:
-                        key_to_value[key] = legacy_results[i]
+                    legacy_value = legacy_results[i]
+                    if legacy_value is not None:
+                        key_to_value[key] = legacy_value
 
                         # Optionally migrate
                         if self.migrate_on_read:
                             (_, ttl) = self.legacy_store.ttl(key=key, collection=collection)
-                            self.current_store.put(key=key, value=legacy_results[i], collection=collection, ttl=ttl)
+                            self.current_store.put(key=key, value=legacy_value, collection=collection, ttl=ttl)
                             if self.delete_after_migration:
                                 self.legacy_store.delete(key=key, collection=collection)
                             self._cache_put(key, collection, "current")
@@ -327,10 +329,10 @@ def ttl_many(self, keys: Sequence[str], *, collection: str | None = None) -> lis
             List of (value, ttl) tuples.
         """
         # Similar logic to get_many but with TTL
-        current_keys = []
-        legacy_keys = []
-        missing_keys = []
-        unknown_keys = []
+        current_keys: list[str] = []
+        legacy_keys: list[str] = []
+        missing_keys: list[str] = []
+        unknown_keys: list[str] = []
 
         for key in keys:
             cached_location = self._cache_get(key, collection)
@@ -343,7 +345,7 @@ def ttl_many(self, keys: Sequence[str], *, collection: str | None = None) -> lis
             else:
                 unknown_keys.append(key)
 
-        key_to_value: dict[str, tuple[dict[str, Any] | None, float | None]] = dict.fromkeys(keys, (None, None))  # type: ignore
+        key_to_value: dict[str, tuple[dict[str, Any] | None, float | None]] = dict.fromkeys(keys, (None, None))
 
         # Fetch from current
         if current_keys:
@@ -368,7 +370,7 @@ def ttl_many(self, keys: Sequence[str], *, collection: str | None = None) -> lis
         if unknown_keys:
             current_results = self.current_store.ttl_many(keys=unknown_keys, collection=collection)
 
-            not_in_current = []
+            not_in_current: list[str] = []
             for i, key in enumerate(unknown_keys):
                 if current_results[i][0] is not None:
                     key_to_value[key] = current_results[i]