数据etl过程

Question

我有以下数据并希望验证它的值( integer、float或string属性)：id度量属性: name属性: cake_name None属性: time属性: time "key‘id“不是字符串，got”键“不是整数，got ”键’点‘不是整数，got “

Answer 1

我的解决方案是一个递归的soltuion，读取嵌套的json数据。

from functools import partial
from typing import Union, Callable
import json

def get_output(key, val, string_keys: list, int_keys: list, float_keys: list):
    out = None
    if key in string_keys:
        if not isinstance(val, str):
            out = f"key '{key}' is not a string, got {type(val)}"
    elif key in int_keys:
        if not isinstance(val, int):
            out = f"key '{key}' is not a integer, got {type(val)}"
    elif key in float_keys:
        if not isinstance(val, float):
            out = f"key '{key}' is not a float, got {type(val)}"
    return out

def explore_json(json: Union[dict, list], validator: Callable):
    result = []
    if isinstance(json, dict):
        for key, val in json.items():
            if isinstance(val, (dict, list)):
                result.extend(explore_json(val, validator))
            else: 
                out = validator(key, val)
                if out is not None:
                    result.append(out) 
    elif isinstance(json, list):
        for val in json:
             result.extend(explore_json(val, validator))
    return result

data = json.loads(json_data)
explore_json(data, validator)
                          
validator = partial(get_output,
                    string_keys=["id", "name", "cake_name", "time"], 
                    int_keys=['metric','points'], 
                    float_keys=["LA:TB2342", "LA:TB2341", "LA:TB2344"])
data = json.loads(json_data)
explore_json(data, validator)

这方面的产出如下：

["key 'id' is not a string, got <class 'NoneType'>",
 "key 'metric' is not a integer, got <class 'str'>",
 "key 'LA:TB2342' is not a float, got <class 'str'>"]

部分函数的优点是，我们可以为每个特定的json提供一个验证器。

此外，请注意，只有在我们的特定验证器中定义的列表string_keys, int_keys, float_keys中的键才能在输出列表中--没有在这些列表中的任何键--才能被验证。

最后，我不确定列表是否与您的相同，但只需更改它们并检查输出。

编辑用于跟踪父键的：

def explore_json(json: Union[dict, list], validator: Callable, parent_key=" parent_key:"):
    result = []
    if isinstance(json, dict):
        for key, val in json.items():
            if isinstance(val, (dict, list)):
                #result = explore_json(val, validator, result)
                result.extend(explore_json(val, validator, f"{parent_key}.{key}"))
            else: 
                out = validator(key, val)
                if out is not None:
                    if parent_key != " parent_key:":
                        out += parent_key
                    result.append(out) 
    elif isinstance(json, list):
        for block_num, val in enumerate(json):
            result.extend(explore_json(val, validator, f"{parent_key}.item{block_num}"))
            # result = explore_json(val, validator, result)
    return result

产出：

["key 'id' is not a string, got <class 'NoneType'>",
 "key 'metric' is not a integer, got <class 'str'>",
 "key 'LA:TB2342' is not a float, got <class 'str'> parent_key:.anticipations.item1.top_properties"]

item1表示该错误位于用于键预期的列表的第一个元素中。