Identify Refactor Opportunities

[strawgate]

We now have a lot more wrappers and a lot more stores.

We should see what functionality is duplicated across these and consider moving them into base classes.

For example, in wrappers perhaps we should have helpers for transforming values and keys and then wrappers can just define those helpers instead of needing to define the whole interface

[claude]

Claude finished @strawgate's task —— View job

---

Refactor Opportunities Analysis Complete ✓

---

Recommendation

Implement a helper-based transformation system in BaseWrapper to reduce boilerplate across wrappers that transform keys, values, or collections.

Many wrappers follow a clear pattern where they:

1. Transform inputs (keys, collections, values) before calling the underlying store
2. Transform outputs (values) after calling the underlying store
3. Pass through operations without modification

Instead of requiring each wrapper to override all 8 protocol methods, wrappers should be able to define simple transformation helpers and have the base class apply them automatically.

---

Findings

Click to expand detailed findings

Current Architecture

The codebase has:

• 14 wrapper implementations that all extend BaseWrapper
• 14 store implementations that extend various Base*Store classes
• BaseStore already provides excellent abstraction by implementing the full protocol based on 3 abstract methods
• BaseWrapper provides a passthrough implementation of all 8 protocol methods

Identified Duplication Patterns in Wrappers

Pattern 1: Key Transformation Wrappers (3 wrappers)

• PrefixKeysWrapper: Adds prefix to keys
• SingleCollectionWrapper: Maps collection+key to prefixed key
• All override 8 methods just to transform keys before/after calling underlying store

# Current approach (repeated ~48 times across these wrappers)
async def get(self, key: str, *, collection: str | None = None) -> dict[str, Any] | None:
    new_key: str = self._prefix_key(key=key)
    return await self.key_value.get(key=new_key, collection=collection)

Pattern 2: Collection Transformation Wrappers (1 wrapper)

• PrefixCollectionsWrapper: Transforms collection names
• Overrides 8 methods to transform collection parameter

Pattern 3: Value Transformation Wrappers (4 wrappers)

• CompressionWrapper: Compresses/decompresses values
• BaseEncryptionWrapper: Encrypts/decrypts values
• LimitSizeWrapper: Filters values by size
• DefaultValueWrapper: Replaces None results with default

Each implements pattern:

async def get(...) -> dict[str, Any] | None:
    value = await self.key_value.get(...)
    return self._transform_value(value)  # Different transform per wrapper

async def put(...) -> None:
    transformed = self._transform_value(value)
    await self.key_value.put(..., value=transformed)

Pattern 4: Operation Wrapping (4 wrappers)

• TimeoutWrapper: Wraps calls with asyncio.wait_for
• RetryWrapper: Wraps calls with retry logic
• LoggingWrapper: Logs before/after operations
• StatisticsWrapper: Tracks metrics around operations

These follow a decorator pattern - wrap the operation itself rather than transform data.

Pattern 5: Routing/Conditional (3 wrappers)

• RoutingWrapper: Routes to different stores
• FallbackWrapper: Falls back on error
• PassthroughCacheWrapper: Checks cache, falls back to primary

These have more complex logic and benefit less from abstraction.

Stores Are Well-Designed

The store base classes are already excellent:

• BaseStore implements full protocol from 3 abstract methods: _get_managed_entry, _put_managed_entry, _delete_managed_entry
• Optional bulk operation overrides: _get_managed_entries, _put_managed_entries, _delete_managed_entries
• Additional mixins for features: BaseEnumerateKeysStore, BaseContextManagerStore, etc.
• Seed data support built into BaseStore

No major refactoring needed for stores - the architecture is solid.

---

Detailed Action Plan

Click to expand implementation plan

Phase 1: Add Helper-Based Wrapper System

1.1 Extend BaseWrapper with optional transformation hooks

Add these optional methods to BaseWrapper:

class BaseWrapper(AsyncKeyValue):
    """Base wrapper with optional transformation hooks."""
    
    key_value: AsyncKeyValue
    
    # Optional hooks - override as needed
    def _transform_key_before(self, key: str, *, collection: str | None = None) -> str:
        """Transform key before passing to underlying store."""
        return key
    
    def _transform_keys_before(self, keys: Sequence[str], *, collection: str | None = None) -> Sequence[str]:
        """Transform keys before passing to underlying store."""
        return [self._transform_key_before(key, collection=collection) for key in keys]
    
    def _transform_collection_before(self, collection: str | None) -> str | None:
        """Transform collection before passing to underlying store."""
        return collection
    
    def _transform_value_before(self, value: Mapping[str, Any], *, key: str, collection: str | None = None) -> Mapping[str, Any]:
        """Transform value before passing to underlying store."""
        return value
    
    def _transform_values_before(self, values: Sequence[Mapping[str, Any]], *, keys: Sequence[str], collection: str | None = None) -> Sequence[Mapping[str, Any]]:
        """Transform values before passing to underlying store."""
        return [self._transform_value_before(value, key=key, collection=collection) for value, key in zip(values, keys, strict=True)]
    
    def _transform_value_after(self, value: dict[str, Any] | None, *, key: str, collection: str | None = None) -> dict[str, Any] | None:
        """Transform value after receiving from underlying store."""
        return value
    
    def _transform_values_after(self, values: list[dict[str, Any] | None], *, keys: Sequence[str], collection: str | None = None) -> list[dict[str, Any] | None]:
        """Transform values after receiving from underlying store."""
        return [self._transform_value_after(value, key=key, collection=collection) for value, key in zip(values, keys, strict=True)]
    
    # Update protocol implementations to use hooks
    @override
    async def get(self, key: str, *, collection: str | None = None) -> dict[str, Any] | None:
        transformed_key = self._transform_key_before(key, collection=collection)
        transformed_collection = self._transform_collection_before(collection)
        value = await self.key_value.get(key=transformed_key, collection=transformed_collection)
        return self._transform_value_after(value, key=key, collection=collection)
    
    # Similar updates for get_many, put, put_many, delete, delete_many, ttl, ttl_many

1.2 Simplify existing wrappers

Refactor wrappers to just define helpers:

# PrefixKeysWrapper becomes:
class PrefixKeysWrapper(BaseWrapper):
    def __init__(self, key_value: AsyncKeyValue, prefix: str):
        self.key_value = key_value
        self.prefix = prefix
    
    def _transform_key_before(self, key: str, *, collection: str | None = None) -> str:
        return prefix_key(prefix=self.prefix, key=key)
    
    # That's it! All 8 methods handled by BaseWrapper

# CompressionWrapper becomes:
class CompressionWrapper(BaseWrapper):
    def __init__(self, key_value: AsyncKeyValue, min_size_to_compress: int = 1024):
        self.key_value = key_value
        self.min_size_to_compress = min_size_to_compress
    
    def _transform_value_before(self, value: Mapping[str, Any], *, key: str, collection: str | None = None) -> Mapping[str, Any]:
        return self._compress_value(dict(value))
    
    def _transform_value_after(self, value: dict[str, Any] | None, *, key: str, collection: str | None = None) -> dict[str, Any] | None:
        return self._decompress_value(value)
    
    # Helper methods stay the same
    def _compress_value(self, value: dict[str, Any]) -> dict[str, Any]: ...
    def _decompress_value(self, value: dict[str, Any] | None) -> dict[str, Any] | None: ...

Benefits:

• Reduces code from ~80 lines to ~20 lines per wrapper
• Eliminates ~672 lines of boilerplate (8 methods × 14 wrappers × 6 lines each)
• Makes wrapper purpose crystal clear
• Maintains backward compatibility (wrappers can still override methods if needed)

Phase 2: Create Abstract Base Classes for Common Wrapper Patterns

2.1 Create specialized base classes

class KeyTransformWrapper(BaseWrapper, ABC):
    """Base class for wrappers that only transform keys."""
    
    @abstractmethod
    def _transform_key(self, key: str) -> str:
        """Transform a single key."""
    
    @abstractmethod  
    def _untransform_key(self, key: str) -> str:
        """Reverse the key transformation."""
    
    # Implement _transform_key_before using _transform_key
    # Optionally override _transform_key_after for keys() operations

class ValueTransformWrapper(BaseWrapper, ABC):
    """Base class for wrappers that transform values bidirectionally."""
    
    @abstractmethod
    def _encode_value(self, value: dict[str, Any]) -> dict[str, Any]:
        """Transform value before storage."""
    
    @abstractmethod
    def _decode_value(self, value: dict[str, Any] | None) -> dict[str, Any] | None:
        """Transform value after retrieval."""
    
    # Implement _transform_value_before/_after using encode/decode

2.2 Update wrappers to use specialized bases

class PrefixKeysWrapper(KeyTransformWrapper):
    def _transform_key(self, key: str) -> str:
        return prefix_key(prefix=self.prefix, key=key)
    
    def _untransform_key(self, key: str) -> str:
        return unprefix_key(prefix=self.prefix, key=key)

class CompressionWrapper(ValueTransformWrapper):
    def _encode_value(self, value: dict[str, Any]) -> dict[str, Any]:
        return self._compress_value(value)
    
    def _decode_value(self, value: dict[str, Any] | None) -> dict[str, Any] | None:
        return self._decompress_value(value)

Phase 3: Testing and Migration

1. Add comprehensive tests for new base wrapper functionality
2. Migrate wrappers one at a time, ensuring tests pass
3. Update documentation with new wrapper patterns
4. Update sync library generation

Phase 4: Consider TTL Transformation Support (Optional)

Some wrappers modify TTL (TTLClampWrapper). Could add:

def _transform_ttl_before(self, ttl: SupportsFloat | None) -> SupportsFloat | None:
    """Transform TTL before passing to underlying store."""
    return ttl

---

Related Issues and Pull Requests

Repository	Issue or PR	Relevance
strawgate/py-key-value	#114 - Add routing wrapper	Recently implemented RoutingWrapper using full method overrides
strawgate/py-key-value	#115 - Add some addition store functionality	Added DefaultValueWrapper, would benefit from helper pattern
strawgate/py-key-value	#49 - Add store middleware	Proposes middleware layer - related architectural discussion
strawgate/py-key-value	PR #122 - feat: add seed support, default values, and JSON serialization wrappers	Merged - added seeding to BaseStore (good base class enhancement example)

Related Files

Repository	File	Relevance	Key Sections
strawgate/py-key-value	base.py	Base wrapper class - would receive helper method additions	9-52
strawgate/py-key-value	prefix_keys/wrapper.py	Example of key transformation pattern	25-77
strawgate/py-key-value	compression/wrapper.py	Example of value transformation pattern	62-149
strawgate/py-key-value	encryption/base.py	Another value transformation example	57-176
strawgate/py-key-value	stores/base.py	Well-designed base store architecture (reference for good patterns)	38-424

---