Example code for using Python's graphlib

complex-toposort.py

"""More complex example with shared dependencies"""    
from graphlib import TopologicalSorter    
from dataclasses import dataclass    
from typing import List    
    
    
@dataclass                      
class Package:               
    name: str     
    depends_on: List[str]          
                                               
                                                  
text_editor = Package("editor", ["ui_lib", "lang_server", "network_lib"])    
lang_server = Package("lang_server", ["gcc", "python"])    
network_lib = Package("network_lib", ["gcc"])    
ui_lib = Package("ui_lib", ["gcc"])    
python = Package("python", ["gcc"])    
gcc = Package("gcc", [])      
                            
ts = TopologicalSorter()                             
ts.add(text_editor.name, *text_editor.depends_on)     
ts.add(lang_server.name, *lang_server.depends_on)    
ts.add(network_lib.name, *network_lib.depends_on)    
ts.add(ui_lib.name, *ui_lib.depends_on)    
ts.add(python.name, *python.depends_on)    
ts.add(gcc.name, *gcc.depends_on)    
    
for node in ts.static_order():    
    print(node)  

cycle-toposort.py

"""Simple cycle demonstration"""                                                                                                          
from graphlib import TopologicalSorter                                                                                                          
from dataclasses import dataclass                                                                                                          
from typing import List                                                                                                          
                                                                                                          
                                                                                                          
@dataclass                                                                                                          
class Package:                                                                                                          
    name: str                                                                                                          
    depends_on: List[str]                                                                                                          
                                                                                                          
                                                                                                          
text_editor = Package("editor", ["lang_server"])                                                                                                          
lang_server = Package("lang_server", ["editor"])                                                                                                          
                                                                                                          
ts = TopologicalSorter()                                                                                                          
ts.add(text_editor.name, *text_editor.depends_on)                                                                                                          
ts.add(lang_server.name, *lang_server.depends_on)                                                                                                          
                                                                                                          
# raises CycleError                                                                                                          
# Traceback (most recent call last):                                                                                                          
#   File "/home/cameron/src/scratch/cycle-toposort.py", line 21, in <module>                                                                                                          
#     for node in ts.static_order():                                                                                                          
#   File "/usr/lib/python3.9/graphlib.py", line 242, in static_order                                                                                                          
#     self.prepare()                                                                                                          
#   File "/usr/lib/python3.9/graphlib.py", line 104, in prepare                                                                                                          
#     raise CycleError(f"nodes are in a cycle", cycle)                                                                                                          
# graphlib.CycleError: ('nodes are in a cycle', ['editor', 'lang_server', 'editor'])                                                                                                          
for node in ts.static_order():                                                                                                          
    print(node) 

parallel-toposort.py

"""Process nodes with multiple worker threads"""
import time
from concurrent.futures import ThreadPoolExecutor
from queue import Queue
from graphlib import TopologicalSorter, CycleError
from dataclasses import dataclass
from typing import List

installed_packages_queue = Queue()


@dataclass
class Package:
    name: str
    depends_on: List[str]


def install_package(package):
    print(f"* Installing package {package}...")
    time.sleep(1)
    installed_packages_queue.put(package)


text_editor = Package("editor", ["ui_lib", "lang_server", "network_lib"])
lang_server = Package("lang_server", ["gcc", "python"])
network_lib = Package("network_lib", ["gcc"])
ui_lib = Package("ui_lib", ["gcc"])
python = Package("python", ["gcc"])
gcc = Package("gcc", [])

ts = TopologicalSorter()
ts.add(text_editor.name, *text_editor.depends_on)
ts.add(lang_server.name, *lang_server.depends_on)
ts.add(network_lib.name, *network_lib.depends_on)
ts.add(ui_lib.name, *ui_lib.depends_on)
ts.add(python.name, *python.depends_on)
ts.add(gcc.name, *gcc.depends_on)

try:
    # no more nodes can be added after calling this!
    ts.prepare()
except CycleError as exc:
    print("Oops!")
    raise exc


with ThreadPoolExecutor() as executor:
    while ts.is_active():
        ready_packages = ts.get_ready()
        print(f"Ready to install {ready_packages}")
        executor.map(install_package, ready_packages)

        installed_package = installed_packages_queue.get()
        ts.done(installed_package)
        print(f"Installed package {installed_package}")

simple-toposort.py

from dataclasses import dataclass    
from typing import List    
      
                
@dataclass          
class Package:      
    name: str                   
    depends_on: List[str]    
      
                                   
text_editor = Package("editor", ["ui_lib"])    
ui_lib = Package("ui_lib", ["gcc"])               
gcc = Package("gcc", [])                        
                        
print(text_editor)    
print(ui_lib)      
print(gcc)    
                              
ts = TopologicalSorter()    
ts.add(text_editor.name, *text_editor.depends_on)    
ts.add(ui_lib.name, *ui_lib.depends_on)               
ts.add(gcc.name, *gcc.depends_on)    
                                    
for node in ts.static_order():    
    print(node)