Python TopologicalSorter: เรียงงานจาก dependency และจับวงจรก่อนเริ่ม

pipeline ที่เขียนเป็นลิสต์ build → test → deploy ดูง่ายจนวันที่เพิ่มงานอิสระ, fan-out หรือ dependency หลายชั้นครับ graphlib.TopologicalSorter ใน standard library รับกราฟแบบ “งานนี้ต้องรอใครบ้าง” แล้วคืนลำดับที่ predecessor มาก่อน พร้อมยก CycleError หากมีวงจร

WHY: dependency คือกติกา ไม่ใช่ตำแหน่งในไฟล์

เมื่อทีมแทรก generate_assets ซึ่งรอ compile แต่ไม่รอ unit_test การแก้ index ในลิสต์เริ่มเปราะ กราฟประกาศความสัมพันธ์โดยตรงและเปิดทางรันงานพร้อมกันภายหลัง เอกสาร Python มีทั้ง static_order() สำหรับลำดับเดียวและ prepare()/get_ready()/done() สำหรับ scheduler แบบขนาน

import unittest
from graphlib import CycleError, TopologicalSorter


def plan(graph):
    return list(TopologicalSorter(graph).static_order())


class TopologicalSorterTests(unittest.TestCase):
    def test_dependencies_come_first(self):
        order = plan({"test": {"build"}, "package": {"build", "test"}, "build": set()})
        self.assertLess(order.index("build"), order.index("test"))
        self.assertLess(order.index("test"), order.index("package"))

    def test_cycle_fails_before_work_starts(self):
        with self.assertRaises(CycleError):
            plan({"a": {"b"}, "b": {"a"}})


if __name__ == "__main__":
    unittest.main()

ลำดับที่ถูกอาจมีหลายแบบ

งานอิสระสลับตำแหน่งกันได้ อย่าเขียน test เทียบลิสต์ทั้งหมดถ้า business rule ไม่บังคับ ให้ตรวจเฉพาะว่า predecessor มาก่อน successor เหมือนตัวอย่าง มิฉะนั้น refactor จาก dict insertion order จะทำให้ test พังทั้งที่แผนยังถูกต้อง

สำหรับงานขนาน เรียก prepare() ก่อนเริ่ม worker แล้วดึงชุด get_ready() เมื่อทำเสร็จต้อง done(node) เพื่อปลดงานถัดไป อย่า mark done ก่อน side effect สำเร็จ และต้องออกแบบ retry/idempotency แยกต่างหาก TopologicalSorter แก้ลำดับ ไม่ได้จัด transaction หรือ rollback ให้

ถ้ากราฟมาจาก input ภายนอก ให้จำกัดจำนวน node/edge และ validate ชื่อก่อน เพราะกราฟใหญ่หรือ dependency ขยะกินหน่วยความจำได้ รวมทั้งรายงาน cycle เป็นเส้นทางที่คนแก้ได้ ไม่ใช่แค่ stack trace

Recap

• model กราฟเป็น node: predecessors
• ใช้ static_order() เมื่อต้องการลำดับง่าย ๆ
• test เฉพาะ constraint เพราะคำตอบถูกมีหลายลำดับ
• cycle detection ไม่แทน retry, rollback หรือ idempotency