0
# Valkey Support
1

2
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.
3

4
## Capabilities
5

6
### Synchronous Valkey Client
7

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

10
```python { .api }
11
class FakeValkey(valkey.Valkey):
12
    def __init__(
13
        self,
14
        host: str = "localhost",
15
        port: int = 6379,
16
        db: int = 0,
17
        password: Optional[str] = None,
18
        socket_timeout: Optional[float] = None,
19
        socket_connect_timeout: Optional[float] = None,
20
        socket_keepalive: bool = False,
21
        socket_keepalive_options: Optional[Dict[str, Any]] = None,
22
        connection_pool: Optional[ConnectionPool] = None,
23
        unix_domain_socket_path: Optional[str] = None,
24
        encoding: str = "utf-8",
25
        encoding_errors: str = "strict",
26
        charset: Optional[str] = None,
27
        errors: Optional[str] = None,
28
        decode_responses: bool = False,
29
        retry_on_timeout: bool = False,
30
        retry_on_error: Optional[List[Exception]] = None,
31
        ssl: bool = False,
32
        ssl_keyfile: Optional[str] = None,
33
        ssl_certfile: Optional[str] = None,
34
        ssl_cert_reqs: Optional[str] = None,
35
        ssl_ca_certs: Optional[str] = None,
36
        ssl_ca_data: Optional[str] = None,
37
        ssl_check_hostname: bool = False,
38
        max_connections: Optional[int] = None,
39
        single_connection_client: bool = False,
40
        health_check_interval: int = 0,
41
        client_name: Optional[str] = None,
42
        username: Optional[str] = None,
43
        retry: Optional[Retry] = None,
44
        redis_connect_func: Optional[Callable] = None,
45
        credential_provider: Optional[CredentialProvider] = None,
46
        protocol: int = 2,
47
        # FakeValkey-specific parameters
48
        server: Optional[FakeServer] = None,
49
        version: VersionType = (7,),
50
        lua_modules: Optional[Dict[str, Any]] = None,
51
        **kwargs
52
    ): ...
53
    
54
    @classmethod
55
    def from_url(
56
        cls,
57
        url: str,
58
        encoding: str = "utf-8",
59
        encoding_errors: str = "strict",
60
        decode_responses: bool = False,
61
        **kwargs
62
    ) -> Self: ...
63
```
64

65
### Asynchronous Valkey Client
66

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

69
```python { .api }
70
class FakeAsyncValkey(valkey.asyncio.Valkey):
71
    def __init__(
72
        self,
73
        host: str = "localhost",
74
        port: int = 6379,
75
        db: int = 0,
76
        password: Optional[str] = None,
77
        socket_timeout: Optional[float] = None,
78
        socket_connect_timeout: Optional[float] = None,
79
        socket_keepalive: bool = False,
80
        socket_keepalive_options: Optional[Dict[str, Any]] = None,
81
        connection_pool: Optional[ConnectionPool] = None,
82
        unix_domain_socket_path: Optional[str] = None,
83
        encoding: str = "utf-8",
84
        encoding_errors: str = "strict",
85
        charset: Optional[str] = None,
86
        errors: Optional[str] = None,
87
        decode_responses: bool = False,
88
        retry_on_timeout: bool = False,
89
        retry_on_error: Optional[List[Exception]] = None,
90
        ssl: bool = False,
91
        ssl_keyfile: Optional[str] = None,
92
        ssl_certfile: Optional[str] = None,
93
        ssl_cert_reqs: Optional[str] = None,
94
        ssl_ca_certs: Optional[str] = None,
95
        ssl_ca_data: Optional[str] = None,
96
        ssl_check_hostname: bool = False,
97
        max_connections: Optional[int] = None,
98
        single_connection_client: bool = False,
99
        health_check_interval: int = 0,
100
        client_name: Optional[str] = None,
101
        username: Optional[str] = None,
102
        retry: Optional[Retry] = None,
103
        redis_connect_func: Optional[Callable] = None,
104
        credential_provider: Optional[CredentialProvider] = None,
105
        protocol: int = 2,
106
        # FakeAsyncValkey-specific parameters
107
        server: Optional[FakeServer] = None,
108
        version: VersionType = (7,),
109
        lua_modules: Optional[Dict[str, Any]] = None,
110
        **kwargs
111
    ): ...
112
    
113
    @classmethod
114
    def from_url(
115
        cls,
116
        url: str,
117
        encoding: str = "utf-8",
118
        encoding_errors: str = "strict",
119
        decode_responses: bool = False,
120
        **kwargs
121
    ) -> Self: ...
122
```
123

124
### Strict Valkey Client
125

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

128
```python { .api }
129
class FakeStrictValkey(valkey.StrictValkey):
130
    def __init__(
131
        self,
132
        server: Optional[FakeServer] = None,
133
        version: VersionType = (7,),
134
        lua_modules: Optional[Dict[str, Any]] = None,
135
        **kwargs
136
    ): ...
137
    
138
    @classmethod
139
    def from_url(
140
        cls,
141
        url: str,
142
        encoding: str = "utf-8",
143
        encoding_errors: str = "strict",
144
        decode_responses: bool = False,
145
        **kwargs
146
    ) -> Self: ...
147
```
148

149
### Valkey-Specific Configuration
150

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

153
```python { .api }
154
# Server type is automatically set to "valkey"
155
def __init__(self, *args, **kwargs):
156
    # Ensures server_type is always "valkey" for Valkey clients
157
    kwargs['server_type'] = 'valkey'
158
    super().__init__(*args, **kwargs)
159
```
160

161
## Usage Examples
162

163
### Basic Valkey Client Usage
164

165
```python
166
import fakeredis
167

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

171
try:
172
    # Create a basic Valkey client
173
    client = fakeredis.FakeValkey()
174
    
175
    # All Redis/Valkey operations work identically
176
    client.set('valkey:test', 'hello valkey')
177
    result = client.get('valkey:test')
178
    print(result.decode())  # 'hello valkey'
179
    
180
    # Complex data structures work the same
181
    client.hset('valkey:user', 'name', 'Alice', 'role', 'admin')
182
    user_data = client.hgetall('valkey:user')
183
    print({k.decode(): v.decode() for k, v in user_data.items()})
184

185
except ImportError:
186
    print("Valkey package not installed. Install with: pip install valkey")
187
```
188

189
### Async Valkey Operations
190

191
```python
192
import asyncio
193
import fakeredis
194

195
async def async_valkey_operations():
196
    try:
197
        # Create async Valkey client
198
        client = fakeredis.FakeAsyncValkey()
199
        
200
        # Async operations
201
        await client.set('async:valkey:key', 'async_value')
202
        result = await client.get('async:valkey:key')
203
        print(f"Async result: {result.decode()}")
204
        
205
        # Pipeline operations
206
        async with client.pipeline() as pipe:
207
            pipe.set('pipe:key1', 'value1')
208
            pipe.set('pipe:key2', 'value2')
209
            pipe.get('pipe:key1')
210
            pipe.get('pipe:key2')
211
            results = await pipe.execute()
212
        
213
        print("Pipeline results:", [r.decode() if r else None for r in results[-2:]])
214
        
215
        # Pub/sub operations
216
        pubsub = client.pubsub()
217
        await pubsub.subscribe('valkey:channel')
218
        
219
        # Publish from another client
220
        publisher = fakeredis.FakeAsyncValkey()
221
        await publisher.publish('valkey:channel', 'Hello Valkey!')
222
        
223
        # Read message
224
        message = await pubsub.get_message(timeout=1.0)
225
        if message and message['type'] == 'message':
226
            print(f"Received: {message['data'].decode()}")
227
        
228
        await pubsub.unsubscribe('valkey:channel')
229
        
230
    except ImportError:
231
        print("Valkey package not installed")
232

233
# Run async example
234
asyncio.run(async_valkey_operations())
235
```
236

237
### Shared Server with Valkey Clients
238

239
```python
240
import fakeredis
241

242
try:
243
    # Create a shared server specifically for Valkey
244
    valkey_server = fakeredis.FakeServer(server_type="valkey")
245
    
246
    # Create multiple Valkey clients sharing the same server
247
    client1 = fakeredis.FakeValkey(server=valkey_server)
248
    client2 = fakeredis.FakeValkey(server=valkey_server)
249
    
250
    # Operations from both clients affect the same data
251
    client1.set('shared:data', 'valkey_data')
252
    result = client2.get('shared:data')
253
    print(f"Shared data: {result.decode()}")
254
    
255
    # Mixed sync and async clients on same server
256
    async_client = fakeredis.FakeAsyncValkey(server=valkey_server)
257
    
258
    async def mixed_operations():
259
        # Async client can see sync client's data
260
        result = await async_client.get('shared:data')
261
        print(f"Async sees: {result.decode()}")
262
        
263
        # Async client writes, sync client reads
264
        await async_client.set('async:written', 'from_async')
265
        sync_result = client1.get('async:written')
266
        print(f"Sync sees: {sync_result.decode()}")
267
    
268
    asyncio.run(mixed_operations())
269
    
270
except ImportError:
271
    print("Valkey package not installed")
272
```
273

274
### URL-based Valkey Configuration
275

276
```python
277
import fakeredis
278

279
try:
280
    # Create Valkey client from URL
281
    client = fakeredis.FakeValkey.from_url(
282
        'redis://localhost:6379/0',  # Note: URL scheme is still 'redis'
283
        decode_responses=True
284
    )
285
    
286
    # With authentication
287
    auth_client = fakeredis.FakeValkey.from_url(
288
        'redis://username:password@localhost:6379/1'
289
    )
290
    
291
    # Test operations
292
    client.set('url_test', 'configured from URL')
293
    result = client.get('url_test')
294
    print(f"URL configured client: {result}")  # Auto-decoded due to decode_responses=True
295
    
296
except ImportError:
297
    print("Valkey package not installed")
298
```
299

300
### Valkey Version Compatibility
301

302
```python
303
import fakeredis
304

305
try:
306
    # Configure for specific Valkey version behavior
307
    valkey_7 = fakeredis.FakeValkey(version=(7, 0))
308
    valkey_8 = fakeredis.FakeValkey(version=(8, 0))
309
    
310
    # Version-specific features will be available based on version
311
    # For example, Redis/Valkey 7.0+ features
312
    valkey_7.set('test_key', 'value', ex=60, get=True)  # GET option in SET
313
    
314
    # Test Valkey-specific server type enforcement
315
    print(f"Server type enforced: valkey")
316
    
317
except ImportError:
318
    print("Valkey package not installed")
319
```
320

321
### Testing with Valkey Clients
322

323
```python
324
import fakeredis
325
import unittest
326

327
class TestValkeyIntegration(unittest.TestCase):
328
    def setUp(self):
329
        try:
330
            # Setup fresh Valkey client for each test
331
            self.client = fakeredis.FakeValkey(decode_responses=True)
332
        except ImportError:
333
            self.skipTest("Valkey package not installed")
334
    
335
    def test_basic_operations(self):
336
        # Test basic set/get
337
        self.client.set('test_key', 'test_value')
338
        result = self.client.get('test_key')
339
        self.assertEqual(result, 'test_value')
340
    
341
    def test_hash_operations(self):
342
        # Test hash operations
343
        self.client.hset('user:1', 'name', 'Alice')
344
        self.client.hset('user:1', 'email', 'alice@example.com')
345
        
346
        name = self.client.hget('user:1', 'name')
347
        self.assertEqual(name, 'Alice')
348
        
349
        all_data = self.client.hgetall('user:1')
350
        self.assertEqual(all_data['name'], 'Alice')
351
        self.assertEqual(all_data['email'], 'alice@example.com')
352
    
353
    def test_list_operations(self):
354
        # Test list operations
355
        self.client.lpush('mylist', 'a', 'b', 'c')
356
        length = self.client.llen('mylist')
357
        self.assertEqual(length, 3)
358
        
359
        items = self.client.lrange('mylist', 0, -1)
360
        self.assertEqual(items, ['c', 'b', 'a'])
361
    
362
    async def test_async_operations(self):
363
        try:
364
            async_client = fakeredis.FakeAsyncValkey(decode_responses=True)
365
            
366
            await async_client.set('async_test', 'async_value')
367
            result = await async_client.get('async_test')
368
            self.assertEqual(result, 'async_value')
369
            
370
        except ImportError:
371
            self.skipTest("Valkey package not installed")
372

373
# Async test runner
374
class AsyncTestValkeyIntegration(unittest.IsolatedAsyncioTestCase):
375
    async def asyncSetUp(self):
376
        try:
377
            self.client = fakeredis.FakeAsyncValkey(decode_responses=True)
378
        except ImportError:
379
            self.skipTest("Valkey package not installed")
380
    
381
    async def test_async_pipeline(self):
382
        async with self.client.pipeline() as pipe:
383
            pipe.set('key1', 'value1')
384
            pipe.set('key2', 'value2')
385
            pipe.get('key1')
386
            pipe.get('key2')
387
            results = await pipe.execute()
388
        
389
        # First two are SET results, last two are GET results
390
        self.assertTrue(results[0])  # SET key1
391
        self.assertTrue(results[1])  # SET key2
392
        self.assertEqual(results[2], 'value1')  # GET key1
393
        self.assertEqual(results[3], 'value2')  # GET key2
394

395
if __name__ == '__main__':
396
    unittest.main()
397
```
398

399
### Migrating from Redis to Valkey Client
400

401
```python
402
import fakeredis
403

404
# Migration helper function
405
def migrate_redis_to_valkey(redis_client_code):
406
    """
407
    Example migration from FakeRedis to FakeValkey
408
    Most code remains identical due to API compatibility
409
    """
410
    
411
    # Before (Redis client)
412
    # redis_client = fakeredis.FakeRedis(decode_responses=True)
413
    
414
    # After (Valkey client) - minimal changes needed
415
    try:
416
        valkey_client = fakeredis.FakeValkey(decode_responses=True)
417
        return valkey_client
418
    except ImportError:
419
        print("Valkey not available, falling back to Redis client")
420
        return fakeredis.FakeRedis(decode_responses=True)
421

422
# Example application code that works with both
423
class CacheManager:
424
    def __init__(self, use_valkey=True):
425
        if use_valkey:
426
            try:
427
                self.client = fakeredis.FakeValkey(decode_responses=True)
428
                self.client_type = "valkey"
429
            except ImportError:
430
                self.client = fakeredis.FakeRedis(decode_responses=True)
431
                self.client_type = "redis"
432
        else:
433
            self.client = fakeredis.FakeRedis(decode_responses=True)
434
            self.client_type = "redis"
435
    
436
    def set_cache(self, key, value, expire=3600):
437
        """Cache operations work identically"""
438
        return self.client.set(key, value, ex=expire)
439
    
440
    def get_cache(self, key):
441
        """Get operations work identically"""
442
        return self.client.get(key)
443
    
444
    def clear_cache_pattern(self, pattern):
445
        """Pattern-based clearing works identically"""
446
        keys = self.client.keys(pattern)
447
        if keys:
448
            return self.client.delete(*keys)
449
        return 0
450
    
451
    def get_info(self):
452
        return f"Using {self.client_type} client"
453

454
# Usage
455
cache = CacheManager(use_valkey=True)
456
print(cache.get_info())
457

458
cache.set_cache('user:123', 'user_data')
459
data = cache.get_cache('user:123')
460
print(f"Cached data: {data}")
461
```
462

463
### Error Handling and Compatibility
464

465
```python
466
import fakeredis
467

468
def create_client_with_fallback():
469
    """Create Valkey client with Redis fallback"""
470
    
471
    try:
472
        # Try to create Valkey client
473
        client = fakeredis.FakeValkey(decode_responses=True)
474
        print("Using FakeValkey client")
475
        return client, "valkey"
476
        
477
    except ImportError:
478
        # Fall back to Redis client
479
        client = fakeredis.FakeRedis(decode_responses=True)
480
        print("Valkey not available, using FakeRedis client")
481
        return client, "redis"
482

483
def test_client_compatibility():
484
    """Test that both clients work identically"""
485
    
486
    client, client_type = create_client_with_fallback()
487
    
488
    # These operations work identically regardless of client type
489
    test_cases = [
490
        ("Basic string", lambda c: c.set('test', 'value') and c.get('test')),
491
        ("Hash ops", lambda c: c.hset('h', 'f', 'v') and c.hget('h', 'f')),
492
        ("List ops", lambda c: c.lpush('l', 'item') and c.llen('l')),
493
        ("Set ops", lambda c: c.sadd('s', 'member') and c.scard('s')),
494
        ("Sorted set ops", lambda c: c.zadd('z', {'m': 1}) and c.zcard('z')),
495
    ]
496
    
497
    print(f"Testing {client_type} client compatibility:")
498
    for test_name, test_func in test_cases:
499
        try:
500
            result = test_func(client)
501
            print(f"  ✓ {test_name}: {'PASS' if result else 'FAIL'}")
502
        except Exception as e:
503
            print(f"  ✗ {test_name}: ERROR - {e}")
504

505
test_client_compatibility()
506
```
507

508
### Installation and Requirements
509

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

512
```bash
513
# Install valkey package for Valkey client support
514
pip install valkey
515

516
# Or install with specific version
517
pip install valkey>=6.0.0
518

519
# FakeRedis will automatically detect and enable Valkey support
520
```
521

522
The Valkey clients in FakeRedis require:
523
- Python >= 3.8 
524
- valkey >= 6.0.0
525
- All the same dependencies as FakeRedis
526

527
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.