攝影或3C

Python遞迴遍歷巢狀資料:return 結束的是目前分支,不是整個 traverse()

在處理 JSON 或設定檔時,資料通常不是只有一層,而是由:

  • dict
  • list
  • 純量,例如 strintboolNone

互相巢狀組成。

例如:

data = {
    "steps": [
        {
            "class_name": "LoginStep",
            "action": {
                "bypass": False
            }
        },
        {
            "class_name": "NetworkStep",
            "action": {
                "bypass": True
            },
            "children": [
                {
                    "class_name": "PingStep",
                    "action": {
                        "bypass": False
                    }
                }
            ]
        }
    ],
    "metadata": {
        "owner": "QA",
        "version": 2
    }
}

我們希望:

  1. 遍歷所有巢狀的 dictlist
  2. 找出具有 class_name 的節點;
  3. 依照 action["bypass"] 分成 Active 與 Bypass;
  4. 同時記錄節點路徑。

一、遞迴的核心概念

遞迴函式就是:

函式在處理目前資料時,再呼叫自己處理子資料。

基本形式如下:

def traverse(node):
    if node 是容器:
        for child in node:
            traverse(child)

遞迴一定要有兩個部分:

1. 繼續條件

遇到可以繼續展開的容器:

dict
list

就繼續呼叫 traverse()

2. 終止條件

遇到沒有子節點的資料:

str
int
float
bool
None

就停止目前這條路徑。

if not isinstance(node, (dict, list)):
    return

這個 return 就是遞迴的終止條件。


二、先看最簡單的遞迴版本

def simple_traverse(node):
    if isinstance(node, dict):
        for key, value in node.items():
            simple_traverse(value)

    elif isinstance(node, list):
        for item in node:
            simple_traverse(item)

    else:
        return

它的邏輯是:

其實純量的 else 也可以省略:

def simple_traverse(node):
    if isinstance(node, dict):
        for value in node.values():
            simple_traverse(value)

    elif isinstance(node, list):
        for item in node:
            simple_traverse(item)

因為當 node 不是 dict 也不是 list 時,函式會自然走到結尾,自動回傳:

return None

但在正式程式中,明確寫出:

if not isinstance(node, (dict, list)):
    return

會更容易表達:

純量是遞迴的終止點。


三、最重要的觀念:每次遞迴都是一次新的函式呼叫

假設程式執行:

traverse(data, "root")

遇到子節點時:

traverse(value, child_path)

這會建立另一個新的 traverse() 呼叫。

可以想成:

 1 次呼叫處理 root
     2 次呼叫處理 root | steps
         3 次呼叫處理 root | steps | 0

每一次呼叫都有自己的:

  • node
  • path
  • 執行位置
  • 回傳位置

所以在某一層執行:

return

只會結束:

目前這一次 traverse() 呼叫。

它會回到上一層,而不是直接消滅所有遞迴層級。


四、return 結束的是分支,不是整個遞迴

請看這個資料:

data = {
    "e": None,
    "a": {
        "b": [1, 2, {"c": 3}],
        "d": "hello"
    }
}

呼叫:

traverse(data, "root")

第一層是字典,因此程式執行:

for key, value in data.items():
    traverse(value, child_path)

先處理:

"e": None

等同於呼叫:

traverse(None, "root | e")

因為 None 是純量:

if not isinstance(None, (dict, list)):
    return

這個 return 的意思是:

結束 traverse(None, "root | e")

然後程式回到上一層:

traverse(data, "root")

回到上一層的 for 迴圈後,繼續處理下一個 key:

"a"

因此後面的資料仍會被遞迴。


執行流程

traverse(root)

├── e: None
│   └── traverse(root | e)
│       └── None 是純量
│           └── return
│               └── 回到 traverse(root)

└── a: dict
    ├── b: list
    │   ├── 1
    │   │   └── 純量return
    │   ├── 2
    │   │   └── 純量return
    │   └── c: 3
    │       └── 純量return

    └── d: "hello"
        └── 純量return

所以:

None  return
只結束 e 這條分支
不會結束 a 這條分支

五、用程式印出實際流程

def traverse(node, path):
    print(f"進入:{path},型別:{type(node).__name__}")

    if not isinstance(node, (dict, list)):
        print(f"離開:{path},原因:純量,return")
        return

    if isinstance(node, dict):
        for key, value in node.items():
            child_path = f"{path} | {key}"
            traverse(value, child_path)

    elif isinstance(node, list):
        for index, item in enumerate(node):
            child_path = f"{path} | [{index}]"
            traverse(item, child_path)

    print(f"完成容器:{path}")

執行:

traverse(data, "root")

可能看到:

注意:

root | e return 之後
root | a 仍然會執行

六、使用閉包保存 Active / Bypass 桶

在實際專案中,bucket 可以由外層函式建立,內層 traverse() 透過閉包使用它們。

from typing import Any


PATH_SEPARATOR = " | "


def collect_nodes(data: Any) -> dict:
    active_bucket = []
    bypassed_bucket = []

    def traverse(
        node: Any,
        node_path: str,
    ) -> None:
        # 純量沒有子節點只結束目前這次 traverse 呼叫
        if not isinstance(node, (dict, list)):
            return

        if isinstance(node, dict):
            if "class_name" in node:
                action = node.get("action", {})

                if isinstance(action, dict):
                    bypass = action.get("bypass", False)
                else:
                    bypass = False

                record = {
                    "class_name": node["class_name"],
                    "node_path": node_path,
                    "bypass": bypass,
                }

                if bypass is True:
                    bypassed_bucket.append(record)
                else:
                    active_bucket.append(record)

            # 繼續處理目前 dict 裡的每一個 value
            for key, value in node.items():
                child_path = (
                    f"{node_path}"
                    f"{PATH_SEPARATOR}{key}"
                )

                traverse(value, child_path)

        elif isinstance(node, list):
            for index, item in enumerate(node):
                child_path = (
                    f"{node_path}"
                    f"{PATH_SEPARATOR}[{index}]"
                )

                traverse(item, child_path)

    traverse(data, "root")

    return {
        "active": active_bucket,
        "bypassed": bypassed_bucket,
    }

七、為什麼這裡適合使用閉包?

active_bucket 和 bypassed_bucket 是外層函式的區域變數:

def collect_nodes(data):
    active_bucket = []
    bypassed_bucket = []

    def traverse(node, node_path):
        ...

雖然 traverse() 沒有接收 bucket 參數,但它仍然可以使用:

active_bucket.append(record)
bypassed_bucket.append(record)

因為內層函式可以捕捉外層函式的變數,這就是閉包。

程式結構可以想成:

collect_nodes()
├── active_bucket
├── bypassed_bucket
└── traverse()
    ├── 使用 active_bucket
    ├── 使用 bypassed_bucket
    └── 遞迴呼叫自己

這種設計下,traverse() 只需要傳遞真正會變動的資訊:

node
node_path

而不需要每次遞迴都重複傳入兩個 bucket。


八、執行結果

result = collect_nodes(data)

print(result["active"])
print(result["bypassed"])

對於前面的資料,結果會類似:

{
    "active": [
        {
            "class_name": "LoginStep",
            "node_path": "root | steps | [0]",
            "bypass": False
        },
        {
            "class_name": "PingStep",
            "node_path": "root | steps | [1] | children | [0]",
            "bypass": False
        }
    ],
    "bypassed": [
        {
            "class_name": "NetworkStep",
            "node_path": "root | steps | [1]",
            "bypass": True
        }
    ]
}

九、目前節點被分桶後,是否還要繼續往下?

這是兩個不同問題:

問題一:目前節點要放哪個 bucket?

if bypass is True:
    bypassed_bucket.append(record)
else:
    active_bucket.append(record)

問題二:目前節點的子節點要不要繼續遞迴?

for key, value in node.items():
    traverse(value, child_path)

在這個教學範例中:

即使目前節點是 Bypass,仍然繼續處理它的子節點。

所以會出現:

NetworkStepBypass
PingStepActive

這是合理的,因為父節點的 bypass 狀態不一定代表所有子節點都必須忽略。


十、如果某個節點需要停止往下走

如果專案規則是:

目前節點一旦 invalid,就不再分析它底下的所有子節點。

可以寫:

if is_invalid(node):
    invalid_bucket.append({
        "node_path": node_path,
        "data": node,
    })
    return

這裡的 return 仍然只會:

停止目前 invalid 節點以下的子樹

不會停止:

  • 父層的其他 key;
  • 同層的其他節點;
  • 其他檔案;
  • 整批資料處理。

例如:

root
├── invalid_node
│   ├── child_A不再處理
│   └── child_B不再處理

└── valid_node
    └── 繼續處理

十一、returnbreakraise 的差異

這三個指令的影響範圍不同。

return
結束目前的函式呼叫:

目前遞迴分支停止
回到上一層

break

for item in items:
    if condition:
        break

結束目前這一層的迴圈:

目前迴圈停止
但不一定離開整個函式

raise

raise ValueError("錯誤")

拋出例外:

如果外層沒有捕捉可能中止整個處理流程

比較如下:

十二、唯一需要補充的例外情況

嚴格來說,如果最外層一開始就是純量:

collect_nodes(None)

那麼:

traverse(None, "root")

會直接 return

這時整個 traversal 看起來就結束了,因為根節點本身沒有任何子節點。

但這是因為:

根節點就是終止點

不是因為某個深層分支的 return 把整棵樹中止了。

比較如下:

深層節點是純量
結束目前分支父層繼續

根節點是純量
沒有任何分支可走因此整次 traversal 結束

十三、遞迴的完整心法

在這類 JSON 遞迴中,可以記住以下順序:

1. 先判斷目前節點是不是容器
2. 不是容器就 return
3.  dict 就處理目前 dict
4. 遍歷 dict 的所有 value
5.  list 就遍歷所有 item
6. 每個子節點都呼叫 traverse()
7. 每次遞迴都攜帶新的路徑
8. 需要時將目前節點放入 Active  Bypass bucket

用程式表示:

def traverse(node, path):
    if not isinstance(node, (dict, list)):
        return

    if isinstance(node, dict):
        process_current_node(node, path)

        for key, value in node.items():
            traverse(
                value,
                make_child_path(path, key),
            )

    elif isinstance(node, list):
        for index, item in enumerate(node):
            traverse(
                item,
                make_child_path(path, index),
            )

最後總結

這段:

if not isinstance(data, (dict, list)):
    return

應該理解成:

目前這次 traverse() 呼叫收到的是純量,
沒有子節點可再展開,因此結束目前這一次呼叫,
回到上一層的迴圈;
上一層會繼續處理其他尚未走訪的子節點。

核心流程是:

dict / list
繼續展開

純量
return結束目前分支

父層迴圈
繼續下一個兄弟節點

因此:

return 目前分支

return 整個遞迴系統

這就是遞迴遍歷巢狀資料時最重要的

推薦hahow線上學習python: https://igrape.net/30afN

儲蓄保險王

儲蓄險是板主最喜愛的儲蓄工具,最喜愛的投資理財工具則是ETF,最喜愛的省錢工具則是信用卡