tessl/pypi-fakeredis

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

—

Pending

Overview

Eval results

Files

Hash Operations

Name: tessl/pypi-fakeredis
Author: tessl

Redis hash data type operations providing a mapping between string fields and string values. Hashes are perfect for representing objects and can hold up to 2^32 - 1 field-value pairs, making them ideal for storing structured data like user profiles, configuration settings, and object attributes.

Capabilities

Basic Hash Operations

Core hash manipulation functions for setting, getting, and managing field-value pairs.

def hset(
    self,
    name: KeyT,
    key: Optional[FieldT] = None,
    value: Optional[EncodableT] = None,
    mapping: Optional[Mapping[AnyFieldT, EncodableT]] = None
) -> ResponseT: ...

def hget(self, name: KeyT, key: FieldT) -> Optional[bytes]: ...

def hmget(self, name: KeyT, keys: Iterable[FieldT]) -> List[Optional[bytes]]: ...

def hmset(self, name: KeyT, mapping: Mapping[AnyFieldT, EncodableT]) -> ResponseT: ...

def hgetall(self, name: KeyT) -> Dict[bytes, bytes]: ...

def hdel(self, name: KeyT, *keys: FieldT) -> ResponseT: ...

def hexists(self, name: KeyT, key: FieldT) -> ResponseT: ...

def hlen(self, name: KeyT) -> ResponseT: ...

Hash Field Operations

Functions for working with hash fields including listing keys, values, and field information.

def hkeys(self, name: KeyT) -> ResponseT: ...

def hvals(self, name: KeyT) -> ResponseT: ...

def hstrlen(self, name: KeyT, key: FieldT) -> ResponseT: ...

def hsetnx(self, name: KeyT, key: FieldT, value: EncodableT) -> ResponseT: ...

def hrandfield(
    self,
    key: KeyT,
    count: Optional[int] = None,
    withvalues: bool = False
) -> Union[Optional[bytes], List[bytes], List[Union[bytes, Tuple[bytes, bytes]]]]: ...

Hash Numeric Operations

Numeric operations for hash fields including increment and decrement operations.

def hincrby(self, name: KeyT, key: FieldT, amount: int = 1) -> ResponseT: ...

def hincrbyfloat(self, name: KeyT, key: FieldT, amount: float = 1.0) -> ResponseT: ...

Hash Scanning

Iterator functions for efficiently traversing hash fields and values.

def hscan(
    self,
    name: KeyT,
    cursor: int = 0,
    match: Optional[PatternT] = None,
    count: Optional[int] = None
) -> ResponseT: ...

def hscan_iter(
    self,
    name: KeyT,
    match: Optional[PatternT] = None,
    count: Optional[int] = None
) -> Iterator[Tuple[bytes, bytes]]: ...

Usage Examples

Basic Hash Operations

import fakeredis

client = fakeredis.FakeRedis()

# Set individual field
client.hset('user:1', 'name', 'Alice')
client.hset('user:1', 'email', 'alice@example.com')
client.hset('user:1', 'age', '30')

# Set multiple fields at once
client.hset('user:2', mapping={
    'name': 'Bob',
    'email': 'bob@example.com', 
    'age': '25'
})

# Get individual field
name = client.hget('user:1', 'name')
print(name.decode())  # 'Alice'

# Get multiple fields
values = client.hmget('user:1', ['name', 'email', 'age'])
for value in values:
    if value:
        print(value.decode())

# Get all fields and values
user_data = client.hgetall('user:1')
for field, value in user_data.items():
    print(f"{field.decode()}: {value.decode()}")

Hash Manipulation

import fakeredis

client = fakeredis.FakeRedis()

# Initialize user profile
client.hset('profile:alice', mapping={
    'name': 'Alice Johnson',
    'department': 'Engineering',
    'level': '5',
    'salary': '95000'
})

# Check if field exists
has_salary = client.hexists('profile:alice', 'salary')
print(has_salary)  # True

# Get number of fields
field_count = client.hlen('profile:alice')
print(field_count)  # 4

# Get all field names
fields = client.hkeys('profile:alice')
for field in fields:
    print(field.decode())

# Get all values
values = client.hvals('profile:alice')
for value in values:
    print(value.decode())

# Delete fields
deleted_count = client.hdel('profile:alice', 'salary', 'level')
print(deleted_count)  # 2

Conditional Operations

import fakeredis

client = fakeredis.FakeRedis()

# Set field only if it doesn't exist
result = client.hsetnx('config:app', 'debug_mode', 'false')
print(result)  # 1 if set, 0 if field already exists

# Try to set again (will fail)
result = client.hsetnx('config:app', 'debug_mode', 'true')
print(result)  # 0 (field already exists)

# Get string length of field value
client.hset('user:profile', 'bio', 'Software engineer with 10 years experience')
bio_length = client.hstrlen('user:profile', 'bio')
print(bio_length)  # Length of the bio string

Numeric Operations

import fakeredis

client = fakeredis.FakeRedis()

# Initialize counters
client.hset('stats:page', mapping={
    'views': '100',
    'likes': '25',
    'rating': '4.5'
})

# Increment integer field
new_views = client.hincrby('stats:page', 'views', 1)
print(new_views)  # 101

# Increment by larger amount
new_views = client.hincrby('stats:page', 'views', 50)
print(new_views)  # 151

# Increment float field
new_rating = client.hincrbyfloat('stats:page', 'rating', 0.1)
print(new_rating)  # 4.6

# Handle non-existent fields (start from 0)
page_shares = client.hincrby('stats:page', 'shares', 1)
print(page_shares)  # 1

Random Field Selection

import fakeredis

client = fakeredis.FakeRedis()

# Setup sample data
client.hset('inventory:store1', mapping={
    'apples': '150',
    'bananas': '200', 
    'oranges': '75',
    'grapes': '300',
    'strawberries': '45'
})

# Get random field
random_field = client.hrandfield('inventory:store1')
print(random_field.decode())  # Random fruit name

# Get multiple random fields
random_fields = client.hrandfield('inventory:store1', count=3)
for field in random_fields:
    print(field.decode())

# Get random fields with values
random_items = client.hrandfield('inventory:store1', count=2, withvalues=True)
for i in range(0, len(random_items), 2):
    field = random_items[i].decode()
    value = random_items[i+1].decode()
    print(f"{field}: {value}")

Hash Scanning

import fakeredis

client = fakeredis.FakeRedis()

# Setup large hash
for i in range(1000):
    client.hset(f'large_hash', f'field_{i}', f'value_{i}')

# Scan all fields
cursor = 0
all_fields = []

while True:
    cursor, fields = client.hscan('large_hash', cursor=cursor, count=100)
    all_fields.extend(fields.items())
    if cursor == 0:
        break

print(f"Total fields scanned: {len(all_fields)}")

# Scan with pattern matching
matching_fields = []
for field, value in client.hscan_iter('large_hash', match='field_1*'):
    matching_fields.append((field.decode(), value.decode()))

print(f"Fields matching pattern: {len(matching_fields)}")

Pattern: Object Storage

import fakeredis
import json

class UserProfile:
    def __init__(self, client, user_id):
        self.client = client
        self.key = f'user:{user_id}'
    
    def set_field(self, field, value):
        return self.client.hset(self.key, field, value)
    
    def get_field(self, field):
        value = self.client.hget(self.key, field)
        return value.decode() if value else None
    
    def set_profile(self, **kwargs):
        return self.client.hset(self.key, mapping=kwargs)
    
    def get_profile(self):
        data = self.client.hgetall(self.key)
        return {k.decode(): v.decode() for k, v in data.items()}
    
    def increment_field(self, field, amount=1):
        return self.client.hincrby(self.key, field, amount)

# Usage
client = fakeredis.FakeRedis()
user = UserProfile(client, '12345')

# Set profile data
user.set_profile(
    name='John Doe',
    email='john@example.com',
    login_count='0',
    last_seen='2024-01-15'
)

# Update individual fields
user.set_field('status', 'active')
user.increment_field('login_count')

# Retrieve profile
profile = user.get_profile()
print(profile)

Pattern: Configuration Management

import fakeredis

class ConfigManager:
    def __init__(self, client, app_name):
        self.client = client
        self.key = f'config:{app_name}'
    
    def set_config(self, **kwargs):
        return self.client.hset(self.key, mapping=kwargs)
    
    def get_config(self, key=None):
        if key:
            value = self.client.hget(self.key, key)
            return value.decode() if value else None
        else:
            config = self.client.hgetall(self.key)
            return {k.decode(): v.decode() for k, v in config.items()}
    
    def update_config(self, **kwargs):
        return self.client.hset(self.key, mapping=kwargs)
    
    def remove_config(self, *keys):
        return self.client.hdel(self.key, *keys)
    
    def has_config(self, key):
        return bool(self.client.hexists(self.key, key))

# Usage
client = fakeredis.FakeRedis()
config = ConfigManager(client, 'myapp')

# Set initial configuration
config.set_config(
    database_url='postgresql://localhost/myapp',
    debug_mode='false',
    max_connections='100',
    timeout='30'
)

# Update specific settings
config.update_config(debug_mode='true', timeout='60')

# Check configuration
if config.has_config('database_url'):
    db_url = config.get_config('database_url')
    print(f"Database URL: {db_url}")

# Get all configuration
all_config = config.get_config()
print("Current configuration:", all_config)

Pattern: Session Storage

import fakeredis
import json
import time

class SessionManager:
    def __init__(self, client, session_id):
        self.client = client
        self.key = f'session:{session_id}'
    
    def set_session_data(self, **kwargs):
        # Add timestamp
        kwargs['last_activity'] = str(int(time.time()))
        return self.client.hset(self.key, mapping=kwargs)
    
    def get_session_data(self, field=None):
        if field:
            value = self.client.hget(self.key, field)
            return value.decode() if value else None
        else:
            data = self.client.hgetall(self.key)
            return {k.decode(): v.decode() for k, v in data.items()}
    
    def touch_session(self):
        return self.client.hset(self.key, 'last_activity', str(int(time.time())))
    
    def increment_counter(self, counter_name):
        return self.client.hincrby(self.key, counter_name, 1)
    
    def destroy_session(self):
        return self.client.delete(self.key)

# Usage
client = fakeredis.FakeRedis()
session = SessionManager(client, 'sess_abc123')

# Initialize session
session.set_session_data(
    user_id='12345',
    username='alice',
    role='user',
    page_views='0'
)

# Update session activity
session.touch_session()
session.increment_counter('page_views')

# Get session info
user_id = session.get_session_data('user_id')
all_data = session.get_session_data()
print(f"User: {user_id}, Session data: {all_data}")

Install with Tessl CLI