Skip to content

Instantly share code, notes, and snippets.

@jcefoli
Last active March 19, 2025 22:42
Show Gist options
  • Save jcefoli/0a256e55d39e949a8d920459f594953d to your computer and use it in GitHub Desktop.
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
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