如何动态解析WebSocket场景下含相似数据的不同JSON Schema？

嘿，你找对方向了——随着WebSocket Schema越来越多，重复写处理函数完全是在做无用功，就算用路由字典也只是换了个方式重复劳动。下面我会一步步带你实现一个配置驱动的通用处理类，彻底解决这个问题：

核心思路 #

把每个Schema的「发送路径」「响应解析规则」「服务标识」都塞进配置文件里，然后用一个通用类干两件事：

1. 发消息时，按配置里的路径生成对应格式的请求
2. 收消息时，自动匹配对应的配置，解析出你需要的字段（比如service_name、color），再统一交给业务逻辑处理

第一步：实现嵌套字典的路径解析工具 #

这是整个机制的核心——我们需要能根据字符串路径（比如properties.color、Item:Product:uid）从嵌套字典里取值/赋值。我写了两个简单的工具函数：

def get_nested_value(data: dict, path: str, separator: str = ".") -> any:
    """从嵌套字典中按路径取值，支持自定义分隔符"""
    keys = path.split(separator)
    current = data
    try:
        for key in keys:
            current = current[key]
        return current
    except (KeyError, TypeError):
        return None

def set_nested_value(data: dict, path: str, value: any, separator: str = ".") -> dict:
    """按路径给嵌套字典赋值，不存在的键会自动创建"""
    keys = path.split(separator)
    current = data
    for key in keys[:-1]:
        if key not in current:
            current[key] = {}
        current = current[key]
    current[keys[-1]] = value
    return data

注意配置里的send_id用的是冒号分隔（比如Product:id），所以调用的时候要指定separator=":"参数。

第二步：构建通用WebSocket处理类 #

这个类会帮你搞定配置加载、消息生成、消息解析的所有脏活：

class WebSocketSchemaHandler:
    def __init__(self, config: dict):
        self.config = config
        # 提前建立「响应顶级键 -> 配置」的映射，加快匹配速度
        self.response_key_map = {}
        for schema_name, schema_config in self.config.items():
            # 从color路径里提取响应的顶级键（比如properties.color的顶级键是properties）
            response_field_path = schema_config["color"]
            top_level_key = response_field_path.split(".")[0]
            self.response_key_map[top_level_key] = schema_config

    def generate_send_message(self, schema_name: str, send_id: str) -> dict:
        """根据配置生成对应Schema的发送消息"""
        schema_config = self.config.get(schema_name)
        if not schema_config:
            raise ValueError(f"Unknown schema: {schema_name}")
        send_path = schema_config["send_id"]
        send_msg = {}
        # 用冒号作为分隔符解析发送路径
        set_nested_value(send_msg, send_path, send_id, separator=":")
        return send_msg

    def handle_received_message(self, received_msg: dict):
        """自动匹配配置并处理收到的消息"""
        # 匹配对应的配置：检查响应的顶级键是否在我们的映射里
        matched_config = None
        for top_key in received_msg.keys():
            if top_key in self.response_key_map:
                matched_config = self.response_key_map[top_key]
                break
        if not matched_config:
            print(f"No matching schema found for message: {received_msg}")
            return
        
        # 解析需要的字段
        service_name = matched_config["name"]
        color = get_nested_value(received_msg, matched_config["color"])
        
        # 调用统一的业务处理函数，不用再写N个处理函数了！
        self._process_message(service_name, color)

    def _process_message(self, service_name: str, color: str):
        """所有Schema的消息都走这个统一的业务处理函数"""
        print(f"Processing message from {service_name}: color is {color}")
        # 这里写你的实际业务逻辑，比如存数据库、触发通知等

第三步：实际使用示例 #

现在来测试一下这个类能不能正常工作：

加载配置 #

# 可以把这个配置存在JSON/YAML文件里，更方便维护
schema_config = {
    "schemaA": {
        "name": "service_ABC",
        "color": "properties.color",
        "send_id": "Product:id"
    },
    "schemaB": {
        "name": "service_DEF",
        "color": "configs.colour",
        "send_id": "Item:Product:uid"
    }
}

handler = WebSocketSchemaHandler(schema_config)

生成发送消息 #

# 生成schemaA的发送消息
send_msg_a = handler.generate_send_message("schemaA", "123")
print(send_msg_a)  # 输出: {"Product": {"id": "123"}}

# 生成schemaB的发送消息
send_msg_b = handler.generate_send_message("schemaB", "123")
print(send_msg_b)  # 输出: {"Item": {"Product": {"uid": "123"}}}

处理收到的响应 #

# 处理schemaA的响应
received_msg_a = {"properties": {"id": {"type": "number"}, "color": "green"}}
handler.handle_received_message(received_msg_a)
# 输出: Processing message from service_ABC: color is green

# 处理schemaB的响应
received_msg_b = {"configs": {"id_number": "123", "type": "int"}, "colour": "green"}
handler.handle_received_message(received_msg_b)
# 输出: Processing message from service_DEF: color is green

扩展优化建议 #

• 如果响应的顶级键不唯一，可以给配置加response_match_rules字段（比如{"required_keys": ["configs", "colour"]}），用更灵活的规则匹配
• 如果需要关联请求和响应，可以在发送消息时生成一个临时uuid，把uuid和schema配置存到映射里，收到响应时根据解析出的id找到对应配置
• 把配置文件单独放在JSON/YAML里，用json.load()或yaml.safe_load()加载，方便团队协作维护

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