CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-fakeredis

Python implementation of redis API, can be used for testing purposes

—

Pending
Overview
Eval results
Files

valkey-support.mddocs/

Valkey Support

FakeRedis provides comprehensive support for Valkey clients through dedicated client classes that maintain full compatibility with the Valkey Python library. Valkey is an open-source fork of Redis that provides identical API compatibility while offering additional features and improvements.

Capabilities

Synchronous Valkey Client

Drop-in replacement for the Valkey synchronous client with all Redis command compatibility.

class FakeValkey(valkey.Valkey):
    def __init__(
        self,
        host: str = "localhost",
        port: int = 6379,
        db: int = 0,
        password: Optional[str] = None,
        socket_timeout: Optional[float] = None,
        socket_connect_timeout: Optional[float] = None,
        socket_keepalive: bool = False,
        socket_keepalive_options: Optional[Dict[str, Any]] = None,
        connection_pool: Optional[ConnectionPool] = None,
        unix_domain_socket_path: Optional[str] = None,
        encoding: str = "utf-8",
        encoding_errors: str = "strict",
        charset: Optional[str] = None,
        errors: Optional[str] = None,
        decode_responses: bool = False,
        retry_on_timeout: bool = False,
        retry_on_error: Optional[List[Exception]] = None,
        ssl: bool = False,
        ssl_keyfile: Optional[str] = None,
        ssl_certfile: Optional[str] = None,
        ssl_cert_reqs: Optional[str] = None,
        ssl_ca_certs: Optional[str] = None,
        ssl_ca_data: Optional[str] = None,
        ssl_check_hostname: bool = False,
        max_connections: Optional[int] = None,
        single_connection_client: bool = False,
        health_check_interval: int = 0,
        client_name: Optional[str] = None,
        username: Optional[str] = None,
        retry: Optional[Retry] = None,
        redis_connect_func: Optional[Callable] = None,
        credential_provider: Optional[CredentialProvider] = None,
        protocol: int = 2,
        # FakeValkey-specific parameters
        server: Optional[FakeServer] = None,
        version: VersionType = (7,),
        lua_modules: Optional[Dict[str, Any]] = None,
        **kwargs
    ): ...
    
    @classmethod
    def from_url(
        cls,
        url: str,
        encoding: str = "utf-8",
        encoding_errors: str = "strict",
        decode_responses: bool = False,
        **kwargs
    ) -> Self: ...

Asynchronous Valkey Client

Async Valkey client with full asyncio support and compatibility with valkey.asyncio.Valkey.

class FakeAsyncValkey(valkey.asyncio.Valkey):
    def __init__(
        self,
        host: str = "localhost",
        port: int = 6379,
        db: int = 0,
        password: Optional[str] = None,
        socket_timeout: Optional[float] = None,
        socket_connect_timeout: Optional[float] = None,
        socket_keepalive: bool = False,
        socket_keepalive_options: Optional[Dict[str, Any]] = None,
        connection_pool: Optional[ConnectionPool] = None,
        unix_domain_socket_path: Optional[str] = None,
        encoding: str = "utf-8",
        encoding_errors: str = "strict",
        charset: Optional[str] = None,
        errors: Optional[str] = None,
        decode_responses: bool = False,
        retry_on_timeout: bool = False,
        retry_on_error: Optional[List[Exception]] = None,
        ssl: bool = False,
        ssl_keyfile: Optional[str] = None,
        ssl_certfile: Optional[str] = None,
        ssl_cert_reqs: Optional[str] = None,
        ssl_ca_certs: Optional[str] = None,
        ssl_ca_data: Optional[str] = None,
        ssl_check_hostname: bool = False,
        max_connections: Optional[int] = None,
        single_connection_client: bool = False,
        health_check_interval: int = 0,
        client_name: Optional[str] = None,
        username: Optional[str] = None,
        retry: Optional[Retry] = None,
        redis_connect_func: Optional[Callable] = None,
        credential_provider: Optional[CredentialProvider] = None,
        protocol: int = 2,
        # FakeAsyncValkey-specific parameters
        server: Optional[FakeServer] = None,
        version: VersionType = (7,),
        lua_modules: Optional[Dict[str, Any]] = None,
        **kwargs
    ): ...
    
    @classmethod
    def from_url(
        cls,
        url: str,
        encoding: str = "utf-8",
        encoding_errors: str = "strict",
        decode_responses: bool = False,
        **kwargs
    ) -> Self: ...

Strict Valkey Client

Backward-compatible strict Valkey client that provides stricter protocol adherence.

class FakeStrictValkey(valkey.StrictValkey):
    def __init__(
        self,
        server: Optional[FakeServer] = None,
        version: VersionType = (7,),
        lua_modules: Optional[Dict[str, Any]] = None,
        **kwargs
    ): ...
    
    @classmethod
    def from_url(
        cls,
        url: str,
        encoding: str = "utf-8",
        encoding_errors: str = "strict",
        decode_responses: bool = False,
        **kwargs
    ) -> Self: ...

Valkey-Specific Configuration

The Valkey clients automatically enforce server_type="valkey" for proper Valkey compatibility mode.

# Server type is automatically set to "valkey"
def __init__(self, *args, **kwargs):
    # Ensures server_type is always "valkey" for Valkey clients
    kwargs['server_type'] = 'valkey'
    super().__init__(*args, **kwargs)

Usage Examples

Basic Valkey Client Usage

import fakeredis

# Note: Requires the 'valkey' package to be installed
# pip install valkey

try:
    # Create a basic Valkey client
    client = fakeredis.FakeValkey()
    
    # All Redis/Valkey operations work identically
    client.set('valkey:test', 'hello valkey')
    result = client.get('valkey:test')
    print(result.decode())  # 'hello valkey'
    
    # Complex data structures work the same
    client.hset('valkey:user', 'name', 'Alice', 'role', 'admin')
    user_data = client.hgetall('valkey:user')
    print({k.decode(): v.decode() for k, v in user_data.items()})

except ImportError:
    print("Valkey package not installed. Install with: pip install valkey")

Async Valkey Operations

import asyncio
import fakeredis

async def async_valkey_operations():
    try:
        # Create async Valkey client
        client = fakeredis.FakeAsyncValkey()
        
        # Async operations
        await client.set('async:valkey:key', 'async_value')
        result = await client.get('async:valkey:key')
        print(f"Async result: {result.decode()}")
        
        # Pipeline operations
        async with client.pipeline() as pipe:
            pipe.set('pipe:key1', 'value1')
            pipe.set('pipe:key2', 'value2')
            pipe.get('pipe:key1')
            pipe.get('pipe:key2')
            results = await pipe.execute()
        
        print("Pipeline results:", [r.decode() if r else None for r in results[-2:]])
        
        # Pub/sub operations
        pubsub = client.pubsub()
        await pubsub.subscribe('valkey:channel')
        
        # Publish from another client
        publisher = fakeredis.FakeAsyncValkey()
        await publisher.publish('valkey:channel', 'Hello Valkey!')
        
        # Read message
        message = await pubsub.get_message(timeout=1.0)
        if message and message['type'] == 'message':
            print(f"Received: {message['data'].decode()}")
        
        await pubsub.unsubscribe('valkey:channel')
        
    except ImportError:
        print("Valkey package not installed")

# Run async example
asyncio.run(async_valkey_operations())

Shared Server with Valkey Clients

import fakeredis

try:
    # Create a shared server specifically for Valkey
    valkey_server = fakeredis.FakeServer(server_type="valkey")
    
    # Create multiple Valkey clients sharing the same server
    client1 = fakeredis.FakeValkey(server=valkey_server)
    client2 = fakeredis.FakeValkey(server=valkey_server)
    
    # Operations from both clients affect the same data
    client1.set('shared:data', 'valkey_data')
    result = client2.get('shared:data')
    print(f"Shared data: {result.decode()}")
    
    # Mixed sync and async clients on same server
    async_client = fakeredis.FakeAsyncValkey(server=valkey_server)
    
    async def mixed_operations():
        # Async client can see sync client's data
        result = await async_client.get('shared:data')
        print(f"Async sees: {result.decode()}")
        
        # Async client writes, sync client reads
        await async_client.set('async:written', 'from_async')
        sync_result = client1.get('async:written')
        print(f"Sync sees: {sync_result.decode()}")
    
    asyncio.run(mixed_operations())
    
except ImportError:
    print("Valkey package not installed")

URL-based Valkey Configuration

import fakeredis

try:
    # Create Valkey client from URL
    client = fakeredis.FakeValkey.from_url(
        'redis://localhost:6379/0',  # Note: URL scheme is still 'redis'
        decode_responses=True
    )
    
    # With authentication
    auth_client = fakeredis.FakeValkey.from_url(
        'redis://username:password@localhost:6379/1'
    )
    
    # Test operations
    client.set('url_test', 'configured from URL')
    result = client.get('url_test')
    print(f"URL configured client: {result}")  # Auto-decoded due to decode_responses=True
    
except ImportError:
    print("Valkey package not installed")

Valkey Version Compatibility

import fakeredis

try:
    # Configure for specific Valkey version behavior
    valkey_7 = fakeredis.FakeValkey(version=(7, 0))
    valkey_8 = fakeredis.FakeValkey(version=(8, 0))
    
    # Version-specific features will be available based on version
    # For example, Redis/Valkey 7.0+ features
    valkey_7.set('test_key', 'value', ex=60, get=True)  # GET option in SET
    
    # Test Valkey-specific server type enforcement
    print(f"Server type enforced: valkey")
    
except ImportError:
    print("Valkey package not installed")

Testing with Valkey Clients

import fakeredis
import unittest

class TestValkeyIntegration(unittest.TestCase):
    def setUp(self):
        try:
            # Setup fresh Valkey client for each test
            self.client = fakeredis.FakeValkey(decode_responses=True)
        except ImportError:
            self.skipTest("Valkey package not installed")
    
    def test_basic_operations(self):
        # Test basic set/get
        self.client.set('test_key', 'test_value')
        result = self.client.get('test_key')
        self.assertEqual(result, 'test_value')
    
    def test_hash_operations(self):
        # Test hash operations
        self.client.hset('user:1', 'name', 'Alice')
        self.client.hset('user:1', 'email', 'alice@example.com')
        
        name = self.client.hget('user:1', 'name')
        self.assertEqual(name, 'Alice')
        
        all_data = self.client.hgetall('user:1')
        self.assertEqual(all_data['name'], 'Alice')
        self.assertEqual(all_data['email'], 'alice@example.com')
    
    def test_list_operations(self):
        # Test list operations
        self.client.lpush('mylist', 'a', 'b', 'c')
        length = self.client.llen('mylist')
        self.assertEqual(length, 3)
        
        items = self.client.lrange('mylist', 0, -1)
        self.assertEqual(items, ['c', 'b', 'a'])
    
    async def test_async_operations(self):
        try:
            async_client = fakeredis.FakeAsyncValkey(decode_responses=True)
            
            await async_client.set('async_test', 'async_value')
            result = await async_client.get('async_test')
            self.assertEqual(result, 'async_value')
            
        except ImportError:
            self.skipTest("Valkey package not installed")

# Async test runner
class AsyncTestValkeyIntegration(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        try:
            self.client = fakeredis.FakeAsyncValkey(decode_responses=True)
        except ImportError:
            self.skipTest("Valkey package not installed")
    
    async def test_async_pipeline(self):
        async with self.client.pipeline() as pipe:
            pipe.set('key1', 'value1')
            pipe.set('key2', 'value2')
            pipe.get('key1')
            pipe.get('key2')
            results = await pipe.execute()
        
        # First two are SET results, last two are GET results
        self.assertTrue(results[0])  # SET key1
        self.assertTrue(results[1])  # SET key2
        self.assertEqual(results[2], 'value1')  # GET key1
        self.assertEqual(results[3], 'value2')  # GET key2

if __name__ == '__main__':
    unittest.main()

Migrating from Redis to Valkey Client

import fakeredis

# Migration helper function
def migrate_redis_to_valkey(redis_client_code):
    """
    Example migration from FakeRedis to FakeValkey
    Most code remains identical due to API compatibility
    """
    
    # Before (Redis client)
    # redis_client = fakeredis.FakeRedis(decode_responses=True)
    
    # After (Valkey client) - minimal changes needed
    try:
        valkey_client = fakeredis.FakeValkey(decode_responses=True)
        return valkey_client
    except ImportError:
        print("Valkey not available, falling back to Redis client")
        return fakeredis.FakeRedis(decode_responses=True)

# Example application code that works with both
class CacheManager:
    def __init__(self, use_valkey=True):
        if use_valkey:
            try:
                self.client = fakeredis.FakeValkey(decode_responses=True)
                self.client_type = "valkey"
            except ImportError:
                self.client = fakeredis.FakeRedis(decode_responses=True)
                self.client_type = "redis"
        else:
            self.client = fakeredis.FakeRedis(decode_responses=True)
            self.client_type = "redis"
    
    def set_cache(self, key, value, expire=3600):
        """Cache operations work identically"""
        return self.client.set(key, value, ex=expire)
    
    def get_cache(self, key):
        """Get operations work identically"""
        return self.client.get(key)
    
    def clear_cache_pattern(self, pattern):
        """Pattern-based clearing works identically"""
        keys = self.client.keys(pattern)
        if keys:
            return self.client.delete(*keys)
        return 0
    
    def get_info(self):
        return f"Using {self.client_type} client"

# Usage
cache = CacheManager(use_valkey=True)
print(cache.get_info())

cache.set_cache('user:123', 'user_data')
data = cache.get_cache('user:123')
print(f"Cached data: {data}")

Error Handling and Compatibility

import fakeredis

def create_client_with_fallback():
    """Create Valkey client with Redis fallback"""
    
    try:
        # Try to create Valkey client
        client = fakeredis.FakeValkey(decode_responses=True)
        print("Using FakeValkey client")
        return client, "valkey"
        
    except ImportError:
        # Fall back to Redis client
        client = fakeredis.FakeRedis(decode_responses=True)
        print("Valkey not available, using FakeRedis client")
        return client, "redis"

def test_client_compatibility():
    """Test that both clients work identically"""
    
    client, client_type = create_client_with_fallback()
    
    # These operations work identically regardless of client type
    test_cases = [
        ("Basic string", lambda c: c.set('test', 'value') and c.get('test')),
        ("Hash ops", lambda c: c.hset('h', 'f', 'v') and c.hget('h', 'f')),
        ("List ops", lambda c: c.lpush('l', 'item') and c.llen('l')),
        ("Set ops", lambda c: c.sadd('s', 'member') and c.scard('s')),
        ("Sorted set ops", lambda c: c.zadd('z', {'m': 1}) and c.zcard('z')),
    ]
    
    print(f"Testing {client_type} client compatibility:")
    for test_name, test_func in test_cases:
        try:
            result = test_func(client)
            print(f"  ✓ {test_name}: {'PASS' if result else 'FAIL'}")
        except Exception as e:
            print(f"  ✗ {test_name}: ERROR - {e}")

test_client_compatibility()

Installation and Requirements

To use FakeRedis with Valkey support, you need to install the Valkey Python package:

# Install valkey package for Valkey client support
pip install valkey

# Or install with specific version
pip install valkey>=6.0.0

# FakeRedis will automatically detect and enable Valkey support

The Valkey clients in FakeRedis require:

  • Python >= 3.8
  • valkey >= 6.0.0
  • All the same dependencies as FakeRedis

If the Valkey package is not available, FakeRedis will raise an ImportError when attempting to create Valkey client instances, but regular FakeRedis clients will continue to work normally.

Install with Tessl CLI

npx tessl i tessl/pypi-fakeredis

docs

bitmap-operations.md

core-clients.md

generic-operations.md

geospatial-operations.md

hash-operations.md

index.md

list-operations.md

lua-scripting.md

pubsub-operations.md

server-management.md

server-operations.md

set-operations.md

sorted-set-operations.md

stack-extensions.md

stream-operations.md

string-operations.md

transaction-operations.md

valkey-support.md

tile.json