125 lines
3.5 KiB
Python
125 lines
3.5 KiB
Python
# python
|
|
import argparse
|
|
import sys
|
|
from typing import Any
|
|
|
|
import redis
|
|
import yaml
|
|
|
|
|
|
def decode_bulk(value: Any):
|
|
if isinstance(value, bytes):
|
|
try:
|
|
return value.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return repr(value)
|
|
if isinstance(value, (list, tuple, set)):
|
|
return type(value)(decode_bulk(v) for v in value)
|
|
if isinstance(value, dict):
|
|
return {decode_bulk(k): decode_bulk(v) for k, v in value.items()}
|
|
return value
|
|
|
|
|
|
def fetch_key(r: redis.Redis, key: bytes) -> tuple[str, Any]:
|
|
decode_bulk(key)
|
|
t = r.type(key)
|
|
if isinstance(t, bytes):
|
|
t = t.decode()
|
|
|
|
if t == "string":
|
|
val = r.get(key)
|
|
return t, decode_bulk(val)
|
|
|
|
if t == "list":
|
|
# LRANGE 0 -1
|
|
vals = r.lrange(key, 0, -1)
|
|
return t, decode_bulk(vals)
|
|
|
|
if t == "set":
|
|
vals = r.smembers(key)
|
|
# sets are unordered; convert to sorted list for stable YAML
|
|
return t, sorted(decode_bulk(vals))
|
|
|
|
if t == "zset":
|
|
# Withscores=True to preserve order/score
|
|
vals = r.zrange(key, 0, -1, withscores=True)
|
|
# Represent as list of {member, score}
|
|
out = [{"member": decode_bulk(m), "score": float(s)} for m, s in vals]
|
|
return t, out
|
|
|
|
if t == "hash":
|
|
vals = r.hgetall(key)
|
|
return t, decode_bulk(vals)
|
|
|
|
# Unknown or none
|
|
return t, None
|
|
|
|
|
|
def dump_redis_to_yaml(
|
|
host: str,
|
|
port: int,
|
|
db: int,
|
|
password: str | None,
|
|
output_path: str,
|
|
match: str | None,
|
|
scan_count: int,
|
|
include_ttl: bool,
|
|
):
|
|
r = redis.Redis(host=host, port=port, db=db, password=password)
|
|
|
|
dump: dict[str, Any] = {"meta": {"host": host, "port": port, "db": db}, "data": {}}
|
|
|
|
cursor = 0
|
|
while True:
|
|
cursor, keys = r.scan(cursor=cursor, match=match, count=scan_count)
|
|
for key in keys:
|
|
key_s = decode_bulk(key)
|
|
t, val = fetch_key(r, key)
|
|
entry = {"type": t, "value": val}
|
|
if include_ttl:
|
|
ttl = r.ttl(key)
|
|
entry["ttl"] = ttl # seconds; -1 no expire, -2 key missing
|
|
dump["data"][key_s] = entry
|
|
if cursor == 0:
|
|
break
|
|
|
|
with open(output_path, "w", encoding="utf-8") as f:
|
|
yaml.safe_dump(dump, f, sort_keys=True, allow_unicode=True)
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description="Dump a Redis DB to a YAML file.")
|
|
parser.add_argument("--host", default="127.0.0.1", help="Redis host")
|
|
parser.add_argument("--port", type=int, default=6379, help="Redis port")
|
|
parser.add_argument("--db", type=int, default=0, help="Redis DB index")
|
|
parser.add_argument("--password", default=None, help="Redis password")
|
|
parser.add_argument("-o", "--output", required=True, help="Output YAML file path")
|
|
parser.add_argument(
|
|
"--match",
|
|
default=None,
|
|
help="Key pattern for SCAN (e.g., 'user:*'). If omitted, all keys are scanned.",
|
|
)
|
|
parser.add_argument(
|
|
"--scan-count", type=int, default=1000, help="Hint for SCAN per iteration (not a limit)."
|
|
)
|
|
parser.add_argument(
|
|
"--include-ttl", action="store_true", help="Include TTL (seconds) for each key."
|
|
)
|
|
|
|
args = parser.parse_args(argv)
|
|
|
|
dump_redis_to_yaml(
|
|
host=args.host,
|
|
port=args.port,
|
|
db=args.db,
|
|
password=args.password,
|
|
output_path=args.output,
|
|
match=args.match,
|
|
scan_count=args.scan_count,
|
|
include_ttl=args.include_ttl,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|