Unnecessary hard dependencies

[karolinepauls]

Pip install causes the following files to be installed. They probably get there via the "pbr" tool.

$ pip install alembic-viz:

...

Installing collected packages: SQLAlchemy, python-editor, MarkupSafe, graphviz, click, Mako, alembic, alembic-viz
  Attempting uninstall: SQLAlchemy                                                             
    Found existing installation: SQLAlchemy 2.0.32
    Uninstalling SQLAlchemy-2.0.32:                                                            
      Successfully uninstalled SQLAlchemy-2.0.32
  Attempting uninstall: MarkupSafe                                                             
    Found existing installation: MarkupSafe 2.1.5
    Uninstalling MarkupSafe-2.1.5:                                                             
      Successfully uninstalled MarkupSafe-2.1.5
  Attempting uninstall: click                                                                  
    Found existing installation: click 8.1.7                                                   
    Uninstalling click-8.1.7:                                                                  
      Successfully uninstalled click-8.1.7                                                     
  Attempting uninstall: Mako                                                                   
    Found existing installation: Mako 1.3.5                                                    
    Uninstalling Mako-1.3.5:                                                                   
      Successfully uninstalled Mako-1.3.5                                                      
  Attempting uninstall: alembic                                                                
    Found existing installation: alembic 1.13.2
    Uninstalling alembic-1.13.2:                                                               
      Successfully uninstalled alembic-1.13.2    

 ...

Successfully installed Mako-1.0.5 MarkupSafe-0.23 SQLAlchemy-1.1.3 alembic-0.8.8 alembic-viz-0.1.0 click-6.6 graphviz-0.5.1 python-editor-1.0.1

dist-info METADATA:

Requires-Dist: Mako (==1.0.5)
Requires-Dist: MarkupSafe (==0.23)
Requires-Dist: SQLAlchemy (==1.1.3)
Requires-Dist: alembic (==0.8.8)
Requires-Dist: click (==6.6)
Requires-Dist: graphviz (==0.5.1)
Requires-Dist: python-editor (==1.0.1)

[PeterSR]

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())