RedisDict is a Python library that offers a convenient and familiar interface for interacting with Redis, treating it as if it were a Python dictionary. Its goal is to help developers write clean, Pythonic code while using Redis as a storage solution for seamless distributed computing. This simple yet powerful library utilizes Redis as a key-value store and supports various data types, including strings, integers, floats, booleans, lists, and dictionaries. Additionally, developers can extend RedisDict to work with custom objects.
The library includes utility functions for more complex use cases such as caching, batching, and more. By leveraging Redis for efficient key-value storage, RedisDict enables high-performance data management, maintaining efficiency even with large datasets and Redis instances.
Redis is an exceptionally fast database when used appropriately. RedisDict leverages Redis for efficient key-value storage, enabling high-performance data management.
pip install redis-dict
>>> from redis_dict import RedisDict
>>> dic = RedisDict()
>>> dic['foo'] = 42
>>> dic['foo']
42
>>> 'foo' in dic
True
>>> dic["baz"] = "hello world"
>>> dic
{'foo': 42, 'baz': 'hello world'}
In Redis our example looks like this.
127.0.0.1:6379> KEYS "*"
1) "main:foo"
2) "main:baz"
127.0.0.1:6379> GET "main:foo"
"int:42"
127.0.0.1:6379> GET "main:baz"
"str:hello world"
Acting as an identifier for your dictionary across different systems, RedisDict employs namespaces for organized data management. When a namespace isn't specified, "main" becomes the default. Thus allowing for data organization across systems and projects with the same redis instance.
This approach also minimizes the risk of key collisions between different applications, preventing hard-to-debug issues. By leveraging namespaces, RedisDict ensures a cleaner and more maintainable data management experience for developers working on multiple projects.
Redis provides a valuable feature that enables keys to expire. RedisDict supports this feature in the following ways:
dic = RedisDict(expire=10)
dic['gone'] = 'in ten seconds'
Or, for a more Pythonic approach, use a timedelta.
from datetime import timedelta
dic = RedisDict(expire=timedelta(minutes=1))
dic['gone'] = 'in a minute'
dic = RedisDict()
seconds = 60
with dic.expire_at(seconds):
dic['gone'] = 'in sixty seconds'
import time
dic = RedisDict(expire=10, preserve_expiration=True)
dic['gone'] = 'in ten seconds'
time.sleep(5)
dic['gone'] = 'gone in 5 seconds'
Efficiently batch your requests using the Pipeline feature, which can be easily utilized with a context manager.
from redis_dict import RedisDict
dic = RedisDict(namespace="example")
# one round trip to redis
with dic.pipeline():
for index in range(100):
dic[str(index)] = index
You can use RedisDict for distributed computing by starting multiple RedisDict instances on different servers or instances that have access to the same Redis instance:
# On server 1
from redis_dict import RedisDict
dic = RedisDict(namespace="example")
dic["foo"] = "bar"
# On server 2
from redis_dict import RedisDict
dic = RedisDict(namespace="example")
print(dic["foo"]) # outputs "bar"
import time
from datetime import timedelta
from redis_dict import RedisDict
def expensive_function(x):
time.sleep(x)
return x * 2
cache = RedisDict(namespace="cache", expire=timedelta(minutes=60))
def cached_expensive_function(x):
if x not in cache:
cache[x] = expensive_function(x)
return cache[x]
start_time = time.time()
print(cached_expensive_function(5)) # Takes around 5 seconds to compute and caches the result.
print(f"Time taken: {time.time() - start_time:.2f} seconds")
start_time = time.time()
print(cached_expensive_function(5)) # Fetches the result from the cache, taking almost no time.
print(f"Time taken: {time.time() - start_time:.2f} seconds")
from redis_dict import RedisDict
# Create a RedisDict instance with a namespace
dic = RedisDict(namespace="example")
# Set key-value pairs
dic["name"] = "John Doe"
dic["age"] = 32
dic["city"] = "Amsterdam"
# Get value by key, from any instance connected to the same redis/namespace
print(dic["name"]) # Output: John Doe
# Update value by key, got a year older
dic["age"] = 33
# Check if key exists
print("name" in dic) # Output: True
print("country" in dic) # Output: False
# Get value with a default value if the key doesn't exist
print(dic.get("country", "NL")) # Output: NL
# Get the length (number of keys) of the RedisDict
print(len(dic)) # Output: 3
# Iterate over keys
for key in dic:
print(key, dic[key])
# Delete a key-value pair
del dic["city"]
# Clear all key-value pairs in the RedisDict
dic.clear()
# Get the length (number of keys) of the RedisDict
print(len(dic)) # Output: 0
# Update RedisDict with multiple key-value pairs
dic.update({"a": 1, "b": 2, "c": 3})
# Use methods of a normal dict
print(list(dic.keys())) # Output: ['a', 'b', 'c']
print(list(dic.values())) # Output: [1, 2, 3]
print(list(dic.items())) # Output: [('a', 1), ('b', 2), ('c', 3)]
# Using pop() and popitem() methods
value = dic.pop("a")
print(value) # Output: 1
key, value = dic.popitem()
print(key, value) # Output: 'c' 3 (example)
# Using setdefault() method
dic.setdefault("d", 4)
print(dic["d"]) # Output: 4
For more advanced examples of RedisDict, please refer to the unit-test files in the repository. All features and functionalities are thoroughly tested in unit tests (here) Or take a look at load test for batching load test. The unit-tests can be as used as a starting point.
RedisDict supports custom type serialization. Here's how to add a new type:
import json
from redis_dict import RedisDict
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
def encode(self) -> str:
return json.dumps(self.__dict__)
@classmethod
def decode(cls, encoded_str: str) -> 'Person':
return cls(**json.loads(encoded_str))
redis_dict = RedisDict()
# Extend redis dict with the new type
redis_dict.extends_type(Person)
# RedisDict can now seamlessly handle Person instances.
person = Person(name="John", age=32)
redis_dict["person1"] = person
result = redis_dict["person1"]
assert result.name == person.name
assert result.age == person.age
>>> from datetime import datetime
>>> redis_dict.extends_type(datetime, datetime.isoformat, datetime.fromisoformat)
>>> redis_dict["now"] = datetime.now()
>>> redis_dict
{'now': datetime.datetime(2024, 10, 14, 18, 41, 53, 493775)}
>>> redis_dict["now"]
datetime.datetime(2024, 10, 14, 18, 41, 53, 493775)
For more information on extending types.
Setup guide for configuring and utilizing encrypted Redis TLS for redis-dict. Setup guide
For storing encrypted data values, it's possible to use extended types. Take a look at this encrypted test.
The RedisDict library includes a comprehensive suite of tests that ensure its correctness and resilience. The test suite covers various data types, edge cases, and error handling scenarios. It also employs the Hypothesis library for property-based testing, which provides fuzz testing to evaluate the implementation
To configure RedisDict using your Redis config.
Configure both the host and port.
dic = RedisDict(host='127.0.0.1', port=6380)
Configuration with a dictionary.
redis_config = {
'host': '127.0.0.1',
'port': 6380,
}
dic = RedisDict(**redis_config)
pip install redis-dict