在Python Django中实现Redis的hmset，解决数据类型报错问题

目标Redis命令 #

我需要在Python Django应用中实现以下Redis命令：

hmset test_template:TEMPLATE_ID test_tags "[{\"key\":\"test_manual_entry_1\",\"value\":\"Some_value_1\"},{\"key\":\"test_manual_entry_2\",\"value\":\"Some_value_2\"}]"

我的尝试代码 #

我编写了RedisUtil类尝试实现，但调用hset/hmset时均报错：

class RedisUtil:
    def hset(self, key_name, data):
        key_name = "test_template:TEMPLATE_ID"
        list_data = [{"key": "test_manual_entry_1", "value": "Some_value1"}, {"key": "test_manual_entry_2", "value": "Some_value2"}]
        data = {"test_tags": [json.dumps(d) for d in list_data]}  # 输出为列表：['{"key": "test_manual_entry_1", "value": "Some_value1"}', '{"key": "test_manual_entry_2", "value": "Some_value2"}']

报错详情 #

两种调用方式都抛出DataError：

1. 直接传入字典：

self.redis_client.hset(key_name, data)
# 错误：redis.exceptions.DataError: Invalid input of type: 'dict'. Convert to a bytes, string, int or float first.

2. 单独传入列表值：

self.redis_client.hset(key_name, "test_tag", data["test_tags"])
# 错误：redis.exceptions.DataError: Invalid input of type: 'list'. Convert to a bytes, string, int or float first.

同时需要处理列表为空的边缘情况。

解决方案 #

Redis哈希表的字段值仅支持字节、字符串、整数或浮点数，不能直接存储列表或字典。正确的做法是将整个列表序列化为单个JSON字符串，而不是逐个元素序列化：

正确实现代码 #

import json
from redis import Redis

class RedisUtil:
    def __init__(self):
        # 初始化Redis客户端，根据你的配置调整参数
        self.redis_client = Redis(host='localhost', port=6379, db=0)

    def save_test_tags(self):
        key_name = "test_template:TEMPLATE_ID"
        list_data = [{"key": "test_manual_entry_1", "value": "Some_value1"}, {"key": "test_manual_entry_2", "value": "Some_value2"}]
        # 将整个列表序列化为单个JSON字符串
        test_tags_str = json.dumps(list_data)
        # 使用hset存储指定字段
        self.redis_client.hset(key_name, "test_tags", test_tags_str)
        
        # 处理空列表的边缘情况
        empty_list = []
        empty_list_str = json.dumps(empty_list)
        self.redis_client.hset(key_name, "empty_test_tags", empty_list_str)

读取时的反序列化 #

后续读取数据时，需要将存储的JSON字符串反序列化为列表：

def get_test_tags(self):
    key_name = "test_template:TEMPLATE_ID"
    test_tags_str = self.redis_client.hget(key_name, "test_tags")
    if test_tags_str:
        return json.loads(test_tags_str)
    return []

补充说明 #

• 把整个列表转成一个JSON字符串，完全符合Redis哈希值的类型要求
• 空列表序列化后是"[]"，可以正常存储和读取，不会触发类型错误
• 如果使用Redis-py 4.0及以上版本，hmset已被弃用，推荐使用hset传入mapping参数批量设置字段：

  # 一次性设置多个哈希字段
  self.redis_client.hset(key_name, mapping={
      "test_tags": test_tags_str,
      "empty_test_tags": empty_list_str
  })
  

内容的提问来源于stack exchange，提问作者Shubham Srivastava