Python處理yaml和嵌套數據結構技巧示例
引言
分享一些用Python處理yaml和嵌套數據結構的的一些技巧，首先從修改yaml格式文件的問題出發(fā)，演變出了各個解決辦法，又從最后的解決辦法中引申出了普適性更強的嵌套數據結構的定位方法。
- 保留注釋修改yaml文件
- 定位嵌套數據結構
- 定位嵌套數據結構2
保留注釋修改yaml文件
yaml比之json文件的其中一個區(qū)別就是可以注釋，這些注釋有時候是很重要的內容，就像代碼中的注釋一樣，如果是手動編輯自然是沒有問題的，那么如何在保留注釋的情況下用代碼修改yaml文件呢？
假設我們要修改的yaml文件如下:
# 主要維護人
name: zhangsan
# 各集群運維人員
cluster1:
node1:
tomcat: user11
cluster2:
node1:
tomcat: user21不保留注釋
為了演示處理yaml的各個方法，這里把不保留注釋的方法也納入到本文了。
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
很顯然，這不是我們要的結果， 那么就淘汰這個方法吧。
此方法只適用于不需要保留注釋的修改。
正則表達式
既然load, dump方法會丟棄注釋，那么用正則表達式不就可以了么，處理文本一定有正則表達式一席之地的。
假設還是將name: zhangsan改成name: wangwu。
def regex1():
pattern = "name:\s+\w+"
pat = re.compile(pattern=pattern)
# 首先匹配到對應的字符串
sub_text = pat.findall(text)[0]
# 根據這個字符串找到在文本的位置
start_index = text.index(sub_text)
# 根據起始位置計算結束位置
end_index = start_index + len(sub_text)
print(start_index, end_index, text[start_index:end_index])
# 將根據索引替換內容
replace_text = "name: wangwu"
new_text = text[:start_index] + replace_text + text[end_index:]
print("="*10)
print(new_text)輸出如下:
8 22 name: zhangsan
==========
# 主要維護人
name: wangwu
# 各集群運維人員
cluster1:
node1:
tomcat: user11
cluster2:
node1:
tomcat: user21
看起來不錯，好像能夠滿足需求，但是這里有一個問題就是，假設修改是cluster2.node1.tomcat的值呢？
因為文本中有兩個tomcat的值，所以只是通過正則表達式不能一擊即中，需要多一些判斷條件，比如首先找到cluster2的起始位置，然后過濾掉小于這個起始位置的索引值，但是如果還有cluster3,cluster4呢？總的來說還 是需要人工的過一遍，然后根據觀察結果來編寫正則表達式，但是這樣太不智能，太不自動了。
此方法適用于比較容器匹配的文本。
語法樹
其實整個文本的數據結構大致如下:
無論是編程語言還是數據文本，如json, yaml, toml都可以得到這樣的語法樹，通過搜索這顆語法樹，我們就能找到對應的鍵值對。
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'))]))]))])
通過yaml.compose方法我們就能得到一顆節(jié)點樹，并且每個節(jié)點會包括該節(jié)點的文本信息，比如起始，終止的文本索引。
通過觀察我們能找到name: zhangsan的兩個節(jié)點, 鍵name是一個ScalarNode節(jié)點, 值zhangsan也是一個ScalarNode, 所以我們可以打印一下看看是否和正則表達式的結果一致。
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é)我們了解到了數據結構可以抽象成一顆語法樹, 那么利用一些樹的搜索算法就可以定位到目標文本了。
這里展示一下不包含列表節(jié)點的搜索算法。
def find_slice(tree: yaml.MappingNode, keys: List[str]) -> Tuple[Tuple[int, int], Tuple[int, int]]:
"""
找到yaml文件中對應鍵值對的索引, 返回一個((key起始索引, key結束索引+1), (value起始索引, value結束索引+1))的元組
暫時只支持鍵值對的尋找.
比如:
>>> 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("沒有找到對應的值")算法核心在于遞歸。
這里的實現并沒有處理列表節(jié)點(SequenceNode)。
假設我們要找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)
輸出如下
# 主要維護人
name: zhangsan# 各集群運維人員
cluster1:
node1:
tomcat: changedcluster2:
node1:
tomcat: user21
上面的算法只能定位key-value類型的數據結構，現在在此優(yōu)化一下，讓其 支持序列。
def find_slice2(tree: yaml.MappingNode, keys: List[str]) -> Tuple[Tuple[int, int], Tuple[int, int]]:
"""
找到yaml文件中對應鍵值對的索引, 返回一個((key起始索引, key結束索引+1), (value起始索引, value結束索引+1))的元組
暫時只支持鍵值對的尋找.
比如:
>>> 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), "錯誤的數據格式"
# 索引可以是負索引, 比如[1,2,3][-1]
if len(tree.value) < abs(target_key):
raise IndexError("索引值大于列表長度")
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("沒有找到對應的值")假設yaml文件如下:
# 用戶列表
users:
- user1: wangwu
- user2: zhangsan
# 集群中間件版本
cluster:
- name: tomcat
version: 9.0.63
- name: nginx
version: 1.21.6def 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)
輸出如下:
# 用戶列表
users:
- user1: wangwu
- user2: zhangsan# 集群中間件版本
cluster:
- name: tomcat
version: 9.0.63
- name: nginx
version: 1.22.0
結果符合預期。
定位嵌套數據結構2
上面介紹了如何定位嵌套的數據結構樹，這一節(jié)介紹一下如何定位較深的樹結構(主要指python字典)。
鏈式調用get
在獲取api數據的時候因為想要的數據結構比較深，用索引會報錯，那么就 需要捕獲異常，這樣很麻煩，并且代碼很冗長，比如:
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}}上面的數據結構很有可能來自同一個api結構，但是數據結構卻不太一樣。
如果直接用索引，就需要捕獲異常，這樣看起來很煩，那么可以利用字典的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", [])
通過給定一個預期的數據空對象，讓get可以一致寫下去。
寫一個遞歸的get
起始在之前的find_slice方法中，我們就發(fā)現遞歸可以比較好的處理這種嵌套的數據結構，我們可以寫一個遞歸處理函數，用來處理很深的數據結構。
假設數據結構如下:
data = {"message": "success", "data": {"data": {"name": "zhangsan", "scores": {"math": {"mid-term": 88, "end-of-term": 90}}}}}
我們的目標就是獲取數據中張三期中數學成績: 88
實現的遞歸調用如下:
def super_get(data: Union[dict, list], keys: List[Union[str, int]]):
assert isinstance(data, dict) or isinstance(data, list), "只支持字典和列表類型"
key = keys[0]
if isinstance(data, list) and isinstance(key, int):
try:
new_data = data[key]
except IndexError as exc:
raise IndexError("索引值大于列表長度") from exc
elif isinstance(data, dict) and isinstance(key, str):
new_data = data.get(key)
else:
raise ValueError(f"數據類型({type(data)})與索引值類型(f{type(key)}不匹配")
if len(keys) == 1:
return new_data
if not isinstance(new_data, dict) and not isinstance(new_data, list):
raise ValueError("找不到對應的值")
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 第三方庫
其實有語法比較強大的庫，比如jq, 但是畢竟多了一個依賴，并且需要一定的學習成本，但是，如果確定自己需要更多的語法，那么可以去安裝一下第三方庫。
總結
如果遇到較深的嵌套，遞歸總能很好的解決，如果實在想不出比較好的算法，那就找個第三方庫吧，調庫嘛，不寒磣。
源碼地址：https://github.com/youerning/blog/tree/master/py_yaml_nested_data
以上就是Python處理yaml和嵌套數據結構技巧示例的詳細內容，更多關于Python處理yaml嵌套數據結構的資料請關注腳本之家其它相關文章！
相關文章
這篇文章主要介紹了Pandas的數據過濾實現，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2021-01-01
這篇文章主要介紹了Python實現采用進度條實時顯示處理進度的方法,涉及Python數學運算結合時間函數顯示進度效果的相關操作技巧,需要的朋友可以參考下2017-12-12
這篇文章主要介紹了通過實例學習Python Excel操作,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-01-01
今天給大家?guī)淼氖顷P于Python爬蟲的相關知識,文章圍繞著Scrapy中Spider的用法展開,文中有非常詳細的介紹及代碼示例,需要的朋友可以參考下2021-06-06
python使用梯度下降和牛頓法尋找Rosenbrock函數最小值實例
這篇文章主要介紹了python使用梯度下降和牛頓法尋找Rosenbrock函數最小值實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-04-04
這篇文章主要介紹了Python中列表與元組的乘法操作,結合簡單實例形式分析了Python中列表、元組的乘法,并附帶分析了字符串的乘法及元組乘法操作的注意事項,需要的朋友可以參考下2018-02-02
python base64 decode incorrect padding錯誤解決方法
這篇文章主要介紹了python base64 decode incorrect padding錯誤解決方法,本文使用把string補齊等號的方法解決了這個錯誤,需要的朋友可以參考下2015-01-01
這篇文章主要為大家詳細介紹了python正則表達式之作業(yè)計算器，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們2016-03-03
這篇文章主要介紹了基于python實現matlab filter函數過程詳解,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-06-06
這篇文章主要介紹了Pycharm IDE的安裝和使用,本文通過圖文并茂的形式給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2020-04-04