Python處理yaml和嵌套數(shù)據(jù)結(jié)構(gòu)技巧示例

更新時(shí)間：2022年06月23日 10:49:29 作者：youerning

這篇文章主要為大家介紹了Python處理yaml和嵌套數(shù)據(jù)結(jié)構(gòu)技巧示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

引言

分享一些用Python處理yaml和嵌套數(shù)據(jù)結(jié)構(gòu)的的一些技巧，首先從修改yaml格式文件的問(wèn)題出發(fā)，演變出了各個(gè)解決辦法，又從最后的解決辦法中引申出了普適性更強(qiáng)的嵌套數(shù)據(jù)結(jié)構(gòu)的定位方法。

保留注釋修改yaml文件
定位嵌套數(shù)據(jù)結(jié)構(gòu)
定位嵌套數(shù)據(jù)結(jié)構(gòu)2

保留注釋修改yaml文件

yaml比之json文件的其中一個(gè)區(qū)別就是可以注釋?zhuān)@些注釋有時(shí)候是很重要的內(nèi)容，就像代碼中的注釋一樣，如果是手動(dòng)編輯自然是沒(méi)有問(wèn)題的，那么如何在保留注釋的情況下用代碼修改yaml文件呢？

假設(shè)我們要修改的yaml文件如下:

# 主要維護(hù)人
name: zhangsan

# 各集群運(yùn)維人員
cluster1:
  node1:
    tomcat: user11

cluster2:
  node1:
    tomcat: user21

不保留注釋

為了演示處理yaml的各個(gè)方法，這里把不保留注釋的方法也納入到本文了。

def ignore_comment():
    data = yaml.load(text, Loader=yaml.Loader)
    data["name"] = "wangwu"
    print(yaml.dump(data))

輸出如下:

cluster1:
node1:
tomcat: user11
cluster2:
node1:
tomcat: user21
name: wangwu

很顯然，這不是我們要的結(jié)果，那么就淘汰這個(gè)方法吧。

此方法只適用于不需要保留注釋的修改。

正則表達(dá)式

既然load, dump方法會(huì)丟棄注釋?zhuān)敲从谜齽t表達(dá)式不就可以了么，處理文本一定有正則表達(dá)式一席之地的。

假設(shè)還是將name: zhangsan改成name: wangwu。

def regex1():
    pattern = "name:\s+\w+"
    pat = re.compile(pattern=pattern)
    # 首先匹配到對(duì)應(yīng)的字符串
    sub_text = pat.findall(text)[0]
    # 根據(jù)這個(gè)字符串找到在文本的位置
    start_index = text.index(sub_text)
    # 根據(jù)起始位置計(jì)算結(jié)束位置
    end_index = start_index + len(sub_text)
    print(start_index, end_index, text[start_index:end_index])

    # 將根據(jù)索引替換內(nèi)容
    replace_text = "name: wangwu"
    new_text = text[:start_index] + replace_text + text[end_index:]
    print("="*10)
    print(new_text)

輸出如下:

8 22 name: zhangsan
==========
# 主要維護(hù)人
name: wangwu
# 各集群運(yùn)維人員
cluster1:
node1:
tomcat: user11
cluster2:
node1:
tomcat: user21

看起來(lái)不錯(cuò)，好像能夠滿(mǎn)足需求，但是這里有一個(gè)問(wèn)題就是，假設(shè)修改是cluster2.node1.tomcat的值呢？

因?yàn)槲谋局杏袃蓚€(gè)tomcat的值，所以只是通過(guò)正則表達(dá)式不能一擊即中，需要多一些判斷條件，比如首先找到cluster2的起始位置，然后過(guò)濾掉小于這個(gè)起始位置的索引值，但是如果還有cluster3,cluster4呢？總的來(lái)說(shuō)還是需要人工的過(guò)一遍，然后根據(jù)觀察結(jié)果來(lái)編寫(xiě)正則表達(dá)式，但是這樣太不智能，太不自動(dòng)了。

此方法適用于比較容器匹配的文本。

語(yǔ)法樹(shù)

其實(shí)整個(gè)文本的數(shù)據(jù)結(jié)構(gòu)大致如下:

無(wú)論是編程語(yǔ)言還是數(shù)據(jù)文本，如json, yaml, toml都可以得到這樣的語(yǔ)法樹(shù)，通過(guò)搜索這顆語(yǔ)法樹(shù)，我們就能找到對(duì)應(yīng)的鍵值對(duì)。

def tree1():
tree = yaml.compose(text)
print(tree)

輸出如下:

MappingNode(tag='tag:yaml.org,2002:map', value=[(ScalarNode(tag='tag:yaml.org,2002:str', value='name'), ScalarNode(tag='tag:yaml.org,2002:str', value='zhangsan')), (ScalarNode(tag='tag:yaml.org,2002:str', value='cluster1'), MappingNode(tag='tag:yaml.org,2002:map', value=[(ScalarNode(tag='tag:yaml.org,2002:str', value='node1'), MappingNode(tag='tag:yaml.org,2002:map', value=[(ScalarNode(tag='tag:yaml.org,2002:str', value='tomcat'), ScalarNode(tag='tag:yaml.org,2002:str', value='user11'))]))])), (ScalarNode(tag='tag:yaml.org,2002:str', value='cluster2'), MappingNode(tag='tag:yaml.org,2002:map', value=[(ScalarNode(tag='tag:yaml.org,2002:str', value='node1'), MappingNode(tag='tag:yaml.org,2002:map', value=[(ScalarNode(tag='tag:yaml.org,2002:str', value='tomcat'), ScalarNode(tag='tag:yaml.org,2002:str', value='user21'))]))]))])

通過(guò)yaml.compose方法我們就能得到一顆節(jié)點(diǎn)樹(shù)，并且每個(gè)節(jié)點(diǎn)會(huì)包括該節(jié)點(diǎn)的文本信息，比如起始，終止的文本索引。

通過(guò)觀察我們能找到name: zhangsan的兩個(gè)節(jié)點(diǎn), 鍵name是一個(gè)ScalarNode節(jié)點(diǎn), 值z(mì)hangsan也是一個(gè)ScalarNode, 所以我們可以打印一下看看是否和正則表達(dá)式的結(jié)果一致。

def tree2():
    tree = yaml.compose(text)
    key_name_node = tree.value[0][0]
    value_name_node = tree.value[0][1]
    print(key_name_node.start_mark.pointer, value_name_node.end_mark.pointer, key_name_node.value, value_name_node.value)

輸出如下:

8 22 name zhangsan

結(jié)果與正則表達(dá)式一致，所以說(shuō)明這種方法可行并且準(zhǔn)確。

得到了修改文本的索引位置，就可以替換了，這里就不再演示了。

此方法適合保留注釋的修改，并且定位嵌套結(jié)構(gòu)較之正則表達(dá)式要簡(jiǎn)單，并且不需要人工介入。

那么如何定位嵌套結(jié)構(gòu)呢？

定位嵌套數(shù)據(jù)結(jié)構(gòu)

從上一節(jié)我們了解到了數(shù)據(jù)結(jié)構(gòu)可以抽象成一顆語(yǔ)法樹(shù), 那么利用一些樹(shù)的搜索算法就可以定位到目標(biāo)文本了。

這里展示一下不包含列表節(jié)點(diǎn)的搜索算法。

def find_slice(tree: yaml.MappingNode, keys: List[str]) -> Tuple[Tuple[int, int], Tuple[int, int]]:
    """
    找到y(tǒng)aml文件中對(duì)應(yīng)鍵值對(duì)的索引, 返回一個(gè)((key起始索引, key結(jié)束索引+1), (value起始索引, value結(jié)束索引+1))的元組
    暫時(shí)只支持鍵值對(duì)的尋找.

    比如:
    >>> find_slice("name: zhangsan", ["name"])
    ((0, 4), (6, 14))
    """
    if isinstance(tree, str):
        tree = yaml.compose(tree, Loader=yaml.Loader)
    assert isinstance(tree, yaml.MappingNode), "未支持的yaml格式"
    target_key = keys[0]
    for node in tree.value:
        if target_key == node[0].value:
            key_node, value_node = node
            if len(keys) == 1:
                key_pointers = (key_node.start_mark.pointer, key_node.end_mark.pointer)
                value_pointers = (value_node.start_mark.pointer, value_node.end_mark.pointer)
                return (key_pointers, value_pointers)

            return find_slice(node[1], keys[1:])

    return ValueError("沒(méi)有找到對(duì)應(yīng)的值")

算法核心在于遞歸。

這里的實(shí)現(xiàn)并沒(méi)有處理列表節(jié)點(diǎn)(SequenceNode)。

假設(shè)我們要找cluster1.node1.tomcat并將其值改成changed, 代碼如下:

def tree3():
    slices = find_slice(text, ["cluster1", "node1", "tomcat"])
    value_start_index, value_end_index = slices[1]
    replace_text = "changed"
    new_text = text[:value_start_index] + replace_text + text[value_end_index:]
    print(new_text)

輸出如下

# 主要維護(hù)人
name: zhangsan

# 各集群運(yùn)維人員
cluster1:
node1:
tomcat: changed

cluster2:
node1:
tomcat: user21

上面的算法只能定位key-value類(lèi)型的數(shù)據(jù)結(jié)構(gòu)，現(xiàn)在在此優(yōu)化一下，讓其支持序列。

def find_slice2(tree: yaml.MappingNode, keys: List[str]) -> Tuple[Tuple[int, int], Tuple[int, int]]:
    """
    找到y(tǒng)aml文件中對(duì)應(yīng)鍵值對(duì)的索引, 返回一個(gè)((key起始索引, key結(jié)束索引+1), (value起始索引, value結(jié)束索引+1))的元組
    暫時(shí)只支持鍵值對(duì)的尋找.

    比如:
    >>> find_slice2("name: zhangsan", ["name"])
    ((0, 4), (6, 14))
    """
    if isinstance(tree, str):
        tree = yaml.compose(tree, Loader=yaml.Loader)
    target_key = keys[0]

    assert isinstance(tree, yaml.MappingNode) or isinstance(tree, yaml.SequenceNode), "未支持的yaml格式"

    ret_key_node = None
    ret_value_node = None
    value_pointers= (-1, -1)

    if isinstance(tree, yaml.SequenceNode):
        assert isinstance(target_key, int), "錯(cuò)誤的數(shù)據(jù)格式"
        # 索引可以是負(fù)索引, 比如[1,2,3][-1]
        if len(tree.value) < abs(target_key):
            raise IndexError("索引值大于列表長(zhǎng)度")

        node = tree.value[target_key]
        if len(keys) > 1:
            return find_slice2(tree.value[target_key], keys[1:])

        if isinstance(node, yaml.MappingNode):
            ret_key_node, ret_value_node = node.value[0]
        else:
            ret_key_node = node

    if isinstance(tree, yaml.MappingNode):
        for node in tree.value:
            if target_key == node[0].value:
                key_node, value_node = node
                if len(keys) > 1:
                    return find_slice2(node[1], keys[1:])
                ret_key_node = key_node
                ret_value_node = value_node

    if ret_key_node:
        key_pointers = (ret_key_node.start_mark.pointer, ret_key_node.end_mark.pointer)

    if ret_value_node:
        value_pointers = (ret_value_node.start_mark.pointer, ret_value_node.end_mark.pointer)

    if ret_key_node:
        return (key_pointers, value_pointers)

    return ValueError("沒(méi)有找到對(duì)應(yīng)的值")

假設(shè)yaml文件如下:

# 用戶(hù)列表
users:
  - user1: wangwu
  - user2: zhangsan

# 集群中間件版本
cluster:
  - name: tomcat
    version: 9.0.63
  - name: nginx
    version: 1.21.6

def tree4():
    slices = find_slice2(text2, ["cluster", 1, "version"])
    value_start_index, value_end_index = slices[1]
    replace_text = "1.22.0"
    new_text = text2[:value_start_index] + replace_text + text2[value_end_index:]
    print(new_text)

輸出如下:

# 用戶(hù)列表
users:
- user1: wangwu
- user2: zhangsan

# 集群中間件版本
cluster:
- name: tomcat
version: 9.0.63
- name: nginx
version: 1.22.0

結(jié)果符合預(yù)期。

定位嵌套數(shù)據(jù)結(jié)構(gòu)2

上面介紹了如何定位嵌套的數(shù)據(jù)結(jié)構(gòu)樹(shù)，這一節(jié)介紹一下如何定位較深的樹(shù)結(jié)構(gòu)(主要指python字典)。

鏈?zhǔn)秸{(diào)用get

在獲取api數(shù)據(jù)的時(shí)候因?yàn)橄胍臄?shù)據(jù)結(jié)構(gòu)比較深，用索引會(huì)報(bào)錯(cuò)，那么就需要捕獲異常，這樣很麻煩，并且代碼很冗長(zhǎng)，比如:

data1 = {"message": "success", "data": {"limit": 0, "offset": 10, "total": 100, "data": ["value1", "value1"]}}
data2 = {"message": "success", "data": None}
data3 = {"message": "success", "data": {"limit": 0, "offset": 10, "total": 100, "data": None}}

上面的數(shù)據(jù)結(jié)構(gòu)很有可能來(lái)自同一個(gè)api結(jié)構(gòu)，但是數(shù)據(jù)結(jié)構(gòu)卻不太一樣。

如果直接用索引，就需要捕獲異常，這樣看起來(lái)很煩，那么可以利用字典的get方法。

ret = data1.get("data", {}).get("data", [])
if ret:
    pass # 做一些操作
if data2.get("data"):
    ret = data2["data"].get("data", [])
ret = data3.get("data", {}).get("data", [])

通過(guò)給定一個(gè)預(yù)期的數(shù)據(jù)空對(duì)象，讓get可以一致寫(xiě)下去。

寫(xiě)一個(gè)遞歸的get

起始在之前的find_slice方法中，我們就發(fā)現(xiàn)遞歸可以比較好的處理這種嵌套的數(shù)據(jù)結(jié)構(gòu)，我們可以寫(xiě)一個(gè)遞歸處理函數(shù)，用來(lái)處理很深的數(shù)據(jù)結(jié)構(gòu)。

假設(shè)數(shù)據(jù)結(jié)構(gòu)如下:

data = {"message": "success", "data": {"data": {"name": "zhangsan", "scores": {"math": {"mid-term": 88, "end-of-term": 90}}}}}

我們的目標(biāo)就是獲取數(shù)據(jù)中張三期中數(shù)學(xué)成績(jī): 88

實(shí)現(xiàn)的遞歸調(diào)用如下:

def super_get(data: Union[dict, list], keys: List[Union[str, int]]):
    assert isinstance(data, dict) or isinstance(data, list), "只支持字典和列表類(lèi)型"
    key = keys[0]

    if isinstance(data, list) and isinstance(key, int):
        try:
            new_data = data[key]
        except IndexError as exc:
            raise IndexError("索引值大于列表長(zhǎng)度") from exc
    elif isinstance(data, dict) and isinstance(key, str):
        new_data = data.get(key)
    else:
        raise ValueError(f"數(shù)據(jù)類(lèi)型({type(data)})與索引值類(lèi)型(f{type(key)}不匹配")

    if len(keys) == 1:
        return new_data

    if not isinstance(new_data, dict) and not isinstance(new_data, list):
        raise ValueError("找不到對(duì)應(yīng)的值")
    return super_get(new_data, keys[1:])

然后執(zhí)行代碼:

def get2():
    data = {"message": "success", "data": {"data": {"zhangsan": {"scores": {"math": {"mid-term": 88, "end-of-term": 90}}}}}}
    print(super_get(data, ["data", "data", "zhangsan", "scores", "math", "mid-term"]))
    # 輸出 88
    data = {"message": "success", "data": {"data": {"zhangsan": {"scores": {"math": [88, 90]}}}}}
    print(super_get(data, ["data", "data", "zhangsan", "scores", "math", 0]))
    # 輸出 88
    data = {"message": "success", "data": {"data": {"zhangsan": {"scores": {"math": [88, 90]}}}}}
    print(super_get(data, ["data", "data", "zhangsan", "scores", "math", -1]))
    # 輸出 90