Terraform module hierarchy

Terraform module hierarchy.py

import networkx as nx
from pyparsing import *
import matplotlib.pyplot as plt
from pathlib import Path
from pprint import pprint
import sys
import re


if len(sys.argv) != 2:
    workspace = Path(".")
else:
    workspace = Path(sys.argv[1])

# Scan and read .tf files
repos = [d for d in workspace.iterdir() if d.is_dir() and d.name.startswith("onscale-terraform-")]

def find_git_sources(tf_file):
    print(f"Scanning {tf_file}")

    with tf_file.open('r') as f:
        content = f.read()

    # Parser for the source attribute
    source_parser = Suppress("source") + Suppress("=") + QuotedString(quoteChar='"')("source_value")

    # Search for sources in content
    sources_found = [tokens['source_value'] for tokens, start, end in source_parser.scanString(content) if "onscale" in tokens.source_value.lower() and "git@" in tokens.source_value]
    print(f"Sources found: {sources_found}")
    # Print the matching content along with the file path
    if sources_found:
        pprint(f"File: {tf_file}\nMatching content: {sources_found}\n{'-'*60}")
  
    return sources_found

git_sources = {}
for repo in repos:
    git_sources[repo.name] = []

    for tf_file in repo.rglob('*.tf'):
        if not any(x in tf_file.parts for x in [".terraform", ".git", ".infracost"]):
            git_sources[repo.name].extend(find_git_sources(tf_file))
# Extract the repository name from the source string
def extract_repo_name(source_string):
    match = re.search(r'OnScale/(.+)\.git', source_string)
    return match.group(1) if match else None

# 3. Create the graph using networkx
G = nx.DiGraph()
for repo, sources in git_sources.items():
    for source in sources:
        target_repo_name = extract_repo_name(source)
        if target_repo_name:  # only add if we could extract a name
            print(f"Adding edge {repo} -> {target_repo_name}")
            G.add_edge(repo, target_repo_name)

G_recur = G.copy()
def get_node_orderings(lvl):
    global G_recur
    leaf_configs = [x for x in G.nodes() if len(list(G.successors(x)))==0]
    
    print(f"## Level {lvl}:")
    
    for l in leaf_configs:
        print(f"* {l}")
        G.remove_node(l)
    if len(list(G.nodes())):
        print("")
        get_node_orderings(lvl + 1)

print(f"## Level 0:")
for repo in repos:
    if repo.name not in list(G.nodes):
        print(f"* {repo.name}")
print("")
get_node_orderings(1)