Skip to content

Instantly share code, notes, and snippets.

@wrgoldstein
Created May 31, 2023 18:00
Show Gist options
  • Save wrgoldstein/a55fa8e1d4382cdf41be56cbf584f99c to your computer and use it in GitHub Desktop.
Save wrgoldstein/a55fa8e1d4382cdf41be56cbf584f99c to your computer and use it in GitHub Desktop.
import pathlib
import collections
import jinja2
import graphlib
class Engine:
current = None
graph = collections.defaultdict(lambda: dict(deps=[], compiled=None, config={}))
env = jinja2.Environment(
loader=jinja2.FileSystemLoader(""),
finalize=lambda x: x if x is not None else ''
)
def ref(node: str):
Engine.graph[Engine.current]["deps"].append(node)
return node
def config(**config: dict):
Engine.graph[Engine.current]["config"] = config
def render(path: pathlib.Path):
Engine.current = path.stem
t = Engine.env.from_string(path.read_text())
Engine.graph[path.stem]["compiled"] = t.render().strip()
def add_macro(path: pathlib.Path):
m = Engine.env.from_string(path.read_text())
macros = [x for x in dir(m.module) if not x.startswith("_")]
for macro in macros:
Engine.env.globals[macro] = getattr(m.module, macro)
env.globals.update(ref=ref, config=config)
def get_sorted():
for macro in pathlib.Path("macros").glob("**/*.sql"):
Engine.add_macro(macro)
for model in pathlib.Path("models").glob("**/*.sql"):
Engine.render(model)
graph = { k:v["deps"] for k,v in Engine.graph.items() }
ts = graphlib.TopologicalSorter(graph)
return list(ts.static_order())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment