Last active
March 19, 2025 22:42
-
-
Save jcefoli/0a256e55d39e949a8d920459f594953d to your computer and use it in GitHub Desktop.
Script to Import/Export data from Valkey/Redis DBs. Very useful to migrate from Redis 7.4 to Valkey (development diverged; Valkey is based off of Redis 7.2.x and the RDB formats are no longer compatible). Assumes Valkey/Redis server is running locally on the default port without any passwords but you can change all of that
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import redis | |
import json | |
# Connect to Valkey/Redis | |
client = redis.Redis(host='127.0.0.1', decode_responses=True) | |
def dump_to_json(output_file="redis_dump.json"): | |
all_data = {} | |
cursor = 0 | |
while True: | |
cursor, keys = client.scan(cursor, count=1000) | |
for key in keys: | |
key_type = client.type(key) | |
if key_type == 'string': | |
all_data[key] = client.get(key) | |
elif key_type == 'hash': | |
all_data[key] = client.hgetall(key) | |
elif key_type == 'list': | |
all_data[key] = client.lrange(key, 0, -1) | |
elif key_type == 'set': | |
all_data[key] = list(client.smembers(key)) | |
elif key_type == 'zset': | |
all_data[key] = client.zrange(key, 0, -1, withscores=True) | |
if cursor == 0: | |
break | |
with open(output_file, "w") as f: | |
json.dump(all_data, f, indent=2) | |
print(f"Dump completed: {output_file}") | |
def load_from_json(input_file="redis_dump.json"): | |
with open(input_file, "r") as f: | |
all_data = json.load(f) | |
for key, value in all_data.items(): | |
if isinstance(value, str): # String | |
client.set(key, value) | |
elif isinstance(value, dict): # Hash | |
for field, field_value in value.items(): | |
client.hset(key, field, field_value) | |
elif isinstance(value, list): | |
if all(isinstance(i, list) and len(i) == 2 for i in value): # ZSet | |
client.zadd(key, dict(value)) | |
else: # List | |
client.rpush(key, *value) | |
elif isinstance(value, set): # Set | |
client.sadd(key, *value) | |
print("Load completed.") | |
# User choice to dump or load data | |
options = { | |
'1': ('Dump data from Valkey/Redis to JSON', dump_to_json), | |
'2': ('Load data from JSON to Valkey/Redis', load_from_json) | |
} | |
print("Choose an option:") | |
for key, (description, _) in options.items(): | |
print(f"{key}. {description}") | |
choice = input("Enter your choice (1/2): ") | |
if choice in options: | |
_, function = options[choice] | |
function() | |
else: | |
print("Invalid choice. Exiting.") | |
exit() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment