tessl/pypi-fakeredis

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

—

Pending

Overview

Eval results

Files

Core Redis Clients

Name: tessl/pypi-fakeredis
Author: tessl

Main client classes that provide drop-in replacements for redis-py clients with full Redis command compatibility and server emulation. These clients support all Redis operations while maintaining complete API compatibility with the redis-py library.

Capabilities

Synchronous Redis Client

The primary Redis client class that inherits from redis.Redis and provides a complete fake Redis implementation.

class FakeRedis(redis.Redis):
    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,
        # FakeRedis-specific parameters
        server: Optional[FakeServer] = None,
        version: VersionType = (7,),
        server_type: ServerType = "redis",
        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 Redis Client

The async Redis client class that inherits from redis.asyncio.Redis and provides full async/await support for all Redis operations.

class FakeAsyncRedis(redis.asyncio.Redis):
    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,
        # FakeAsyncRedis-specific parameters
        server: Optional[FakeServer] = None,
        version: VersionType = (7,),
        server_type: ServerType = "redis",
        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 Redis Client

Backward-compatible Redis client that provides stricter Redis protocol adherence.

class FakeStrictRedis(redis.StrictRedis):
    def __init__(
        self,
        server: Optional[FakeServer] = None,
        version: VersionType = (7,),
        server_type: ServerType = "redis",
        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: ...

Connection Classes

Low-level connection classes that handle the Redis protocol communication.

class FakeConnection(redis.Connection):
    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,
        socket_type: int = 0,
        redis_connect_func: Optional[Callable] = None,
        encoder: Optional[Encoder] = None,
        # FakeConnection-specific parameters
        server: Optional[FakeServer] = None,
        **kwargs
    ): ...
    
    def connect(self) -> None: ...
    def disconnect(self) -> None: ...
    def can_read(self, timeout: Optional[float] = 0) -> bool: ...
    def read_response(self, **kwargs: Any) -> Any: ...
    def send_command(self, *args: Any) -> None: ...
    def send_packed_command(self, command: bytes) -> None: ...

class FakeAsyncConnection(redis.asyncio.Connection):
    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,
        socket_type: int = 0,
        redis_connect_func: Optional[Callable] = None,
        encoder: Optional[Encoder] = None,
        # FakeAsyncConnection-specific parameters
        server: Optional[FakeServer] = None,
        **kwargs
    ): ...
    
    async def connect(self) -> None: ...
    async def disconnect(self) -> None: ...
    async def can_read(self, timeout: Optional[float] = 0) -> bool: ...
    async def read_response(self, **kwargs: Any) -> Any: ...
    async def send_command(self, *args: Any) -> None: ...
    async def send_packed_command(self, command: bytes) -> None: ...

Usage Examples

Basic Client Usage

import fakeredis

# Create a basic FakeRedis client
client = fakeredis.FakeRedis()

# Perform operations
client.set('key', 'value')
result = client.get('key')
print(result.decode())  # 'value'

Shared Server Instance

import fakeredis

# Create a shared server
server = fakeredis.FakeServer()

# Create multiple clients sharing the same server
client1 = fakeredis.FakeRedis(server=server)
client2 = fakeredis.FakeRedis(server=server)

# Operations from both clients affect the same data
client1.set('shared_key', 'value')
result = client2.get('shared_key')  # Same server, same data

Async Client Usage

import asyncio
import fakeredis

async def async_operations():
    # Create async client
    client = fakeredis.FakeAsyncRedis()
    
    # Perform async operations
    await client.set('async_key', 'async_value')
    result = await client.get('async_key')
    print(result.decode())  # 'async_value'
    
    # Use with context manager
    async with client.pipeline() as pipe:
        pipe.set('key1', 'value1')
        pipe.set('key2', 'value2')
        results = await pipe.execute()

# Run the async function
asyncio.run(async_operations())

URL-based Configuration

import fakeredis

# Create client from URL (works with both sync and async)
client = fakeredis.FakeRedis.from_url(
    'redis://localhost:6379/0',
    decode_responses=True
)

# Redis URL with authentication  
authenticated_client = fakeredis.FakeRedis.from_url(
    'redis://username:password@localhost:6379/1'
)

Version and Server Type Configuration

import fakeredis

# Configure for specific Redis version
redis_6_client = fakeredis.FakeRedis(version=(6, 2))
redis_7_client = fakeredis.FakeRedis(version=(7, 0))

# Configure for different server types
dragonfly_client = fakeredis.FakeRedis(server_type="dragonfly")
valkey_client = fakeredis.FakeRedis(server_type="valkey")  # If valkey support available

Custom Lua Modules

import fakeredis

# Load custom Lua modules
lua_modules = {
    'mymodule': '''
        local function my_function(key, value)
            redis.call("SET", key, value)
            return "OK"
        end
        return {my_function = my_function}
    '''
}

client = fakeredis.FakeRedis(lua_modules=lua_modules)

Install with Tessl CLI