Unnecessary hard dependencies #4

Activity

1. PeterSR commented on Jan 16, 2026

   

   PeterSR

   on Jan 16, 2026

   More actions

   I ended up ditching this package for that reason. Instead I am just parsing the output from alembic history -i.

   Here's my script:

   #!/usr/bin/env python3
   """
   Convert `alembic history -i` output (from stdin or a file) into Graphviz DOT.
   
   Usage:
     alembic history -i | python3 alembic_history_to_dot.py > alembic.dot
     dot -Tsvg alembic.dot -o alembic.svg
   
   Parsing rules (based on Alembic's single-line history format):
     <parent_rev>[, <parent_rev>...] -> <child_rev> [(flag[, flag...]) ...], <message>
   
   - Multiple parents (merge points) are supported via comma-separated LHS.
   - Flags are parsed ONLY from parenthesis groups that appear BEFORE the first comma
     that starts the message (so parentheses in the message don't get misread).
   - Revision identifiers in Alembic are strings; we accept common patterns, not just hex.
   """
   
   from __future__ import annotations
   
   import argparse
   import re
   import sys
   from typing import Dict, Iterable, List, Optional, Set, Tuple, TypedDict
   
   # --- Types ---
   
   Edge = Tuple[str, str]
   
   
   class NodeInfo(TypedDict):
       msg: str
       flags: Set[str]
   
   
   # --- Parsing helpers ---
   
   # Alembic revision ids are "string identifiers". Commonly hex, but not required.
   # We accept:
   #   - <base>
   #   - any non-empty token without whitespace, comma, or parentheses
   REV_ID_RE = re.compile(r"^(<[^>]+>|[^,\s()]+)$")
   
   # Extract "child_rev" as the first token, stopping before whitespace/comma/paren.
   CHILD_TOKEN_RE = re.compile(r"^(?P<rev><[^>]+>|[^,\s()]+)(?P<rest>.*)$")
   
   
   def _split_first_comma_outside_parens(s: str) -> Tuple[str, str]:
       """
       Split s into (before_comma, after_comma) at the first comma not inside (...) groups.
       If no such comma exists, returns (s, "").
       """
       depth = 0
       for i, ch in enumerate(s):
           if ch == "(":
               depth += 1
           elif ch == ")" and depth > 0:
               depth -= 1
           elif ch == "," and depth == 0:
               return s[:i], s[i + 1 :]
       return s, ""
   
   
   def _parse_flags(meta: str) -> Set[str]:
       """
       Parse flags from parenthesis groups, e.g. "(head, current) (branchpoint)".
       Tokens are split on commas and whitespace, normalized to lowercase.
       """
       flags: Set[str] = set()
       for inner in re.findall(r"\(([^)]*)\)", meta):
           for tok in re.split(r"[\s,]+", inner.strip()):
               tok = tok.strip().lower()
               if tok:
                   flags.add(tok)
       return flags
   
   
   def _parse_history_line(line: str) -> Optional[Tuple[List[str], str, Set[str], str]]:
       """
       Parse one history line into (parents, child, flags, msg).
       Returns None if the line doesn't look like an Alembic history arrow line.
       """
       line = line.strip()
       if not line or "->" not in line:
           return None
       if line.startswith("INFO "):
           return None
   
       left, right = (part.strip() for part in line.split("->", 1))
   
       # Parents: comma-separated tokens on the LHS
       parents = [p.strip() for p in left.split(",") if p.strip()]
       if not parents:
           return None
       # Filter/validate parent tokens (keeps parsing resilient if the line has junk)
       parents = [p for p in parents if REV_ID_RE.match(p)]
       if not parents:
           return None
   
       # Child id is the first token on the RHS
       m = CHILD_TOKEN_RE.match(right)
       if not m:
           return None
       child = m.group("rev").strip()
       if not REV_ID_RE.match(child):
           return None
   
       rest = m.group("rest").strip()
   
       # The first comma outside parentheses separates metadata (flags) from the message.
       meta, msg = _split_first_comma_outside_parens(rest)
       meta = meta.strip()
       msg = msg.strip()
   
       flags = _parse_flags(meta)
   
       return parents, child, flags, msg
   
   
   def parse_history(lines: Iterable[str]) -> Tuple[Dict[str, NodeInfo], Set[Edge]]:
       nodes: Dict[str, NodeInfo] = {}
       edges: Set[Edge] = set()
   
       for raw in lines:
           parsed = _parse_history_line(raw)
           if not parsed:
               continue
           parents, child, flags, msg = parsed
   
           for pid in parents:
               nodes.setdefault(pid, {"msg": "", "flags": set()})
   
           nodes.setdefault(child, {"msg": "", "flags": set()})
   
           if msg and not nodes[child]["msg"]:
               nodes[child]["msg"] = msg
           nodes[child]["flags"].update(flags)
   
           for pid in parents:
               edges.add((pid, child))
   
       return nodes, edges
   
   
   # --- DOT emission helpers ---
   
   def _escape_dot(s: str) -> str:
       """Escape for DOT quoted strings."""
       return s.replace("\\", "\\\\").replace('"', '\\"')
   
   
   def _label_value(node_id: str, msg: str) -> str:
       """
       Build an escaped label *value* (no surrounding quotes),
       using DOT's newline escape sequence (\n).
       """
       if msg:
           return f"{_escape_dot(node_id)}\\n{_escape_dot(msg)}"
       return _escape_dot(node_id)
   
   
   def emit_dot(nodes: Dict[str, NodeInfo], edges: Set[Edge], rankdir: str = "LR") -> str:
       out: List[str] = []
       out.append("strict digraph alembic_history {")
       out.append(f"  rankdir={rankdir};")
       out.append('  node [fontname="Helvetica", shape=box];')
       out.append('  edge [fontname="Helvetica"];')
   
       for node_id in sorted(nodes.keys()):
           info = nodes[node_id]
           flags = set(info.get("flags", set()))
           msg = str(info.get("msg", ""))
   
           attrs: Dict[str, str] = {}
           attrs["label"] = _label_value(node_id, msg)
   
           if node_id.lower() == "<base>":
               attrs["shape"] = "oval"
           if "head" in flags:
               attrs["peripheries"] = "2"
           if "branchpoint" in flags:
               attrs["shape"] = "diamond"
           if "mergepoint" in flags:
               attrs["shape"] = "octagon"
           if "current" in flags:
               attrs["style"] = "filled"
   
           attr_str = ", ".join(
               f'{k}="{v if k == "label" else _escape_dot(v)}"'
               for k, v in attrs.items()
           )
           out.append(f'  "{_escape_dot(node_id)}" [{attr_str}];')
   
       for parent, child in sorted(edges):
           out.append(f'  "{_escape_dot(parent)}" -> "{_escape_dot(child)}";')
   
       out.append("}")
       return "\n".join(out) + "\n"
   
   
   def main() -> int:
       ap = argparse.ArgumentParser()
       ap.add_argument(
           "-f", "--file",
           help="Read alembic history output from a file instead of stdin.",
       )
       ap.add_argument(
           "--rankdir",
           default="LR",
           choices=["LR", "TB", "RL", "BT"],
           help="Graphviz rank direction.",
       )
       args = ap.parse_args()
   
       if args.file:
           with open(args.file, "r", encoding="utf-8") as f:
               nodes, edges = parse_history(f)
       else:
           nodes, edges = parse_history(sys.stdin)
   
       sys.stdout.write(emit_dot(nodes, edges, rankdir=args.rankdir))
       return 0
   
   
   if __name__ == "__main__":
       raise SystemExit(main())

   🎉React with 🎉1Reacted by Matěj Šmíd