tessl/pypi-fakeredis

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

—

Pending

Overview

Eval results

Files

Server Management

Name: tessl/pypi-fakeredis
Author: tessl

Server instances that manage Redis state, databases, and connections with support for different Redis-compatible server types. FakeServer provides the central coordination for all Redis operations and maintains consistency across multiple clients.

Capabilities

Core Server Management

Central server instance that manages databases, script cache, pub/sub subscribers, and connection state.

class FakeServer:
    def __init__(
        self,
        version: VersionType = (7,),
        server_type: ServerType = "redis", 
        config: Optional[Dict[bytes, bytes]] = None
    ): ...
    
    @property
    def connected(self) -> bool: ...
    
    @property
    def lock(self) -> threading.RLock: ...
    
    def get_server(self) -> FakeServer: ...
    def get_next_client_id(self) -> int: ...
    def flush_all(self) -> None: ...
    def save(self) -> None: ...
    def shutdown(self) -> None: ...

TCP Server Implementation

Full TCP server implementation for network testing with Redis protocol support and multi-threaded connection handling.

class TcpFakeServer(ThreadingTCPServer):
    def __init__(
        self,
        server_address: Tuple[str, int],
        server_type: ServerType = "redis",
        server_version: VersionType = (8, 0),
        RequestHandlerClass: type = StreamRequestHandler
    ): ...
    
    def server_bind(self) -> None: ...
    def server_activate(self) -> None: ...
    def shutdown(self) -> None: ...
    def get_server(self) -> FakeServer: ...

Usage Examples

Basic Server Usage

import fakeredis

# Create a server with default settings (Redis 7.0)
server = fakeredis.FakeServer()

# Create clients that share the same server
client1 = fakeredis.FakeRedis(server=server)
client2 = fakeredis.FakeRedis(server=server)

# Both clients operate on the same data
client1.set('shared_key', 'value')
result = client2.get('shared_key')  # Returns b'value'

Version Configuration

import fakeredis

# Configure for specific Redis versions
redis_6_server = fakeredis.FakeServer(version=(6, 2, 7))
redis_7_server = fakeredis.FakeServer(version=(7, 0, 5))
redis_8_server = fakeredis.FakeServer(version=(8, 0))

# Clients inherit server version behavior
client_v6 = fakeredis.FakeRedis(server=redis_6_server)
client_v7 = fakeredis.FakeRedis(server=redis_7_server)

Server Type Configuration

import fakeredis

# Configure for different server types
redis_server = fakeredis.FakeServer(server_type="redis")
dragonfly_server = fakeredis.FakeServer(server_type="dragonfly")
valkey_server = fakeredis.FakeServer(server_type="valkey")

# Server type affects command behavior and compatibility
dragonfly_client = fakeredis.FakeRedis(server=dragonfly_server)
# Now supports Dragonfly-specific commands like SADDEX

Server Configuration

import fakeredis

# Configure server with Redis-style config
config = {
    b'requirepass': b'mypassword',
    b'maxmemory': b'1gb',
    b'maxmemory-policy': b'allkeys-lru',
    b'save': b'900 1 300 10'
}

server = fakeredis.FakeServer(config=config)
client = fakeredis.FakeRedis(server=server)

# Client must authenticate
try:
    client.set('key', 'value')  # Will fail without auth
except redis.AuthenticationError:
    pass

# Authenticate first
client.auth('mypassword')
client.set('key', 'value')  # Now works

TCP Server for Network Testing

import fakeredis
import redis
import threading
import time

# Start a TCP server on localhost:6380
tcp_server = fakeredis.TcpFakeServer(
    ('localhost', 6380),
    server_type="redis", 
    server_version=(7, 0)
)

# Run server in background thread
server_thread = threading.Thread(target=tcp_server.serve_forever)
server_thread.daemon = True
server_thread.start()

# Connect with regular redis-py client
real_client = redis.Redis(host='localhost', port=6380, decode_responses=True)

# Test network operations
real_client.set('network_key', 'network_value')
result = real_client.get('network_key')
print(f"Result from TCP server: {result}")

# Shutdown server
tcp_server.shutdown()

Multiple Isolated Servers

import fakeredis

# Create isolated servers for different test scenarios
test_server_1 = fakeredis.FakeServer()
test_server_2 = fakeredis.FakeServer()

# Each server maintains separate data
client_1a = fakeredis.FakeRedis(server=test_server_1)
client_1b = fakeredis.FakeRedis(server=test_server_1)  # Shares with 1a

client_2a = fakeredis.FakeRedis(server=test_server_2)  # Isolated from server 1

# Operations on different servers are independent
client_1a.set('key', 'value1')
client_2a.set('key', 'value2')

print(client_1b.get('key').decode())  # 'value1' (same server as 1a)
print(client_2a.get('key').decode())  # 'value2' (different server)

Server State Management

import fakeredis

server = fakeredis.FakeServer()
client = fakeredis.FakeRedis(server=server)

# Add some data
client.set('key1', 'value1')
client.set('key2', 'value2')
client.lpush('list', 'item1', 'item2')

# Check server connection state
print(f"Server connected: {server.connected}")  # True

# Flush all data
server.flush_all()
print(f"Keys after flush: {client.keys()}")  # []

# Shutdown server (disconnects all clients)
server.shutdown()
print(f"Server connected: {server.connected}")  # False

Server with Custom Database Count

import fakeredis

# Create server with specific database configuration
config = {
    b'databases': b'32'  # Support 32 databases (default is 16)
}

server = fakeredis.FakeServer(config=config)

# Create clients for different databases
client_db0 = fakeredis.FakeRedis(server=server, db=0)
client_db15 = fakeredis.FakeRedis(server=server, db=15)
client_db31 = fakeredis.FakeRedis(server=server, db=31)

# Each database is isolated
client_db0.set('key', 'db0_value')
client_db15.set('key', 'db15_value')

print(client_db0.get('key').decode())   # 'db0_value'
print(client_db15.get('key').decode())  # 'db15_value'
print(client_db31.get('key'))           # None (not set)

Install with Tessl CLI