Add Elixir language support (.ex/.exs)

Extracts defmodule, def/defp, alias/import/require/use, and call graph.
Follows same custom-walk pattern as Zig and PowerShell extractors.

Author: safishamsi

graphify/detect.py

@@ -17,7 +17,7 @@ class FileType(str, Enum):
 
 _MANIFEST_PATH = "graphify-out/manifest.json"
 
-CODE_EXTENSIONS = {'.py', '.ts', '.js', '.tsx', '.go', '.rs', '.java', '.cpp', '.cc', '.cxx', '.c', '.h', '.hpp', '.rb', '.swift', '.kt', '.kts', '.cs', '.scala', '.php', '.lua', '.toc', '.zig', '.ps1'}
+CODE_EXTENSIONS = {'.py', '.ts', '.js', '.tsx', '.go', '.rs', '.java', '.cpp', '.cc', '.cxx', '.c', '.h', '.hpp', '.rb', '.swift', '.kt', '.kts', '.cs', '.scala', '.php', '.lua', '.toc', '.zig', '.ps1', '.ex', '.exs'}
 DOC_EXTENSIONS = {'.md', '.txt', '.rst'}
 PAPER_EXTENSIONS = {'.pdf'}
 IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg'}

graphify/extract.py

@@ -1928,6 +1928,183 @@ def walk_imports(node) -> None:
     return new_edges
 
 
+def extract_elixir(path: Path) -> dict:
+    """Extract modules, functions, imports, and calls from a .ex/.exs file."""
+    try:
+        import tree_sitter_elixir as tselixir
+        from tree_sitter import Language, Parser
+    except ImportError:
+        return {"nodes": [], "edges": [], "error": "tree_sitter_elixir not installed"}
+
+    try:
+        language = Language(tselixir.language())
+        parser = Parser(language)
+        source = path.read_bytes()
+        tree = parser.parse(source)
+        root = tree.root_node
+    except Exception as e:
+        return {"nodes": [], "edges": [], "error": str(e)}
+
+    stem = path.stem
+    str_path = str(path)
+    nodes: list[dict] = []
+    edges: list[dict] = []
+    seen_ids: set[str] = set()
+    function_bodies: list[tuple[str, Any]] = []
+
+    def add_node(nid: str, label: str, line: int) -> None:
+        if nid not in seen_ids:
+            seen_ids.add(nid)
+            nodes.append({"id": nid, "label": label, "file_type": "code",
+                          "source_file": str_path, "source_location": f"L{line}"})
+
+    def add_edge(src: str, tgt: str, relation: str, line: int,
+                 confidence: str = "EXTRACTED", weight: float = 1.0) -> None:
+        edges.append({"source": src, "target": tgt, "relation": relation,
+                      "confidence": confidence, "source_file": str_path,
+                      "source_location": f"L{line}", "weight": weight})
+
+    file_nid = _make_id(stem)
+    add_node(file_nid, path.name, 1)
+
+    _IMPORT_KEYWORDS = frozenset({"alias", "import", "require", "use"})
+
+    def _get_alias_text(node) -> str | None:
+        for child in node.children:
+            if child.type == "alias":
+                return source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
+        return None
+
+    def walk(node, parent_module_nid: str | None = None) -> None:
+        if node.type != "call":
+            for child in node.children:
+                walk(child, parent_module_nid)
+            return
+
+        identifier_node = None
+        arguments_node = None
+        do_block_node = None
+        for child in node.children:
+            if child.type == "identifier":
+                identifier_node = child
+            elif child.type == "arguments":
+                arguments_node = child
+            elif child.type == "do_block":
+                do_block_node = child
+
+        if identifier_node is None:
+            for child in node.children:
+                walk(child, parent_module_nid)
+            return
+
+        keyword = source[identifier_node.start_byte:identifier_node.end_byte].decode("utf-8", errors="replace")
+        line = node.start_point[0] + 1
+
+        if keyword == "defmodule":
+            module_name = _get_alias_text(arguments_node) if arguments_node else None
+            if not module_name:
+                return
+            module_nid = _make_id(stem, module_name)
+            add_node(module_nid, module_name, line)
+            add_edge(file_nid, module_nid, "contains", line)
+            if do_block_node:
+                for child in do_block_node.children:
+                    walk(child, parent_module_nid=module_nid)
+            return
+
+        if keyword in ("def", "defp"):
+            func_name = None
+            if arguments_node:
+                for child in arguments_node.children:
+                    if child.type == "call":
+                        for sub in child.children:
+                            if sub.type == "identifier":
+                                func_name = source[sub.start_byte:sub.end_byte].decode("utf-8", errors="replace")
+                                break
+                    elif child.type == "identifier":
+                        func_name = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
+                        break
+            if not func_name:
+                return
+            container = parent_module_nid or file_nid
+            func_nid = _make_id(container, func_name)
+            add_node(func_nid, f"{func_name}()", line)
+            if parent_module_nid:
+                add_edge(parent_module_nid, func_nid, "method", line)
+            else:
+                add_edge(file_nid, func_nid, "contains", line)
+            if do_block_node:
+                function_bodies.append((func_nid, do_block_node))
+            return
+
+        if keyword in _IMPORT_KEYWORDS and arguments_node:
+            module_name = _get_alias_text(arguments_node)
+            if module_name:
+                tgt_nid = _make_id(module_name)
+                add_edge(file_nid, tgt_nid, "imports", line)
+            return
+
+        for child in node.children:
+            walk(child, parent_module_nid)
+
+    walk(root)
+
+    label_to_nid: dict[str, str] = {}
+    for n in nodes:
+        normalised = n["label"].strip("()").lstrip(".")
+        label_to_nid[normalised.lower()] = n["id"]
+
+    seen_call_pairs: set[tuple[str, str]] = set()
+    _SKIP_KEYWORDS = frozenset({
+        "def", "defp", "defmodule", "defmacro", "defmacrop",
+        "defstruct", "defprotocol", "defimpl", "defguard",
+        "alias", "import", "require", "use",
+        "if", "unless", "case", "cond", "with", "for",
+    })
+
+    def walk_calls(node, caller_nid: str) -> None:
+        if node.type != "call":
+            for child in node.children:
+                walk_calls(child, caller_nid)
+            return
+        for child in node.children:
+            if child.type == "identifier":
+                kw = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
+                if kw in _SKIP_KEYWORDS:
+                    for c in node.children:
+                        walk_calls(c, caller_nid)
+                    return
+                break
+        callee_name: str | None = None
+        for child in node.children:
+            if child.type == "dot":
+                dot_text = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
+                parts = dot_text.rstrip(".").split(".")
+                if parts:
+                    callee_name = parts[-1]
+                break
+            if child.type == "identifier":
+                callee_name = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
+                break
+        if callee_name:
+            tgt_nid = label_to_nid.get(callee_name.lower())
+            if tgt_nid and tgt_nid != caller_nid:
+                pair = (caller_nid, tgt_nid)
+                if pair not in seen_call_pairs:
+                    seen_call_pairs.add(pair)
+                    add_edge(caller_nid, tgt_nid, "calls",
+                             node.start_point[0] + 1, confidence="INFERRED", weight=0.8)
+        for child in node.children:
+            walk_calls(child, caller_nid)
+
+    for caller_nid, body in function_bodies:
+        walk_calls(body, caller_nid)
+
+    clean_edges = [e for e in edges if e["source"] in seen_ids and
+                   (e["target"] in seen_ids or e["relation"] == "imports")]
+    return {"nodes": nodes, "edges": clean_edges, "input_tokens": 0, "output_tokens": 0}
+
+
 # ── Main extract and collect_files ────────────────────────────────────────────
 
 def extract(paths: list[Path]) -> dict:
@@ -1980,6 +2157,8 @@ def extract(paths: list[Path]) -> dict:
         ".toc": extract_lua,
         ".zig": extract_zig,
         ".ps1": extract_powershell,
+        ".ex": extract_elixir,
+        ".exs": extract_elixir,
     }
 
     for path in paths:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "graphifyy"
-version = "0.3.5"
+version = "0.3.6"
 description = "AI coding assistant skill (Claude Code, Codex, OpenCode, OpenClaw) - turn any folder of code, docs, papers, or images into a queryable knowledge graph"
 readme = "README.md"
 license = { file = "LICENSE" }
@@ -30,6 +30,7 @@ dependencies = [
     "tree-sitter-lua",
     "tree-sitter-zig",
     "tree-sitter-powershell",
+    "tree-sitter-elixir",
 ]
 
 [project.urls]

tests/fixtures/sample.ex

@@ -0,0 +1,28 @@
+defmodule MyApp.Accounts.User do
+  @moduledoc """
+  Handles user accounts and authentication.
+  """
+
+  alias MyApp.Repo
+  import Ecto.Query
+
+  defstruct [:id, :name, :email]
+
+  def create(attrs) do
+    %__MODULE__{}
+    |> validate(attrs)
+    |> Repo.insert()
+  end
+
+  def find(id) do
+    Repo.get(__MODULE__, id)
+  end
+
+  defp validate(user, attrs) do
+    if Map.has_key?(attrs, :email) do
+      user
+    else
+      {:error, :missing_email}
+    end
+  end
+end

tests/test_extract.py

@@ -61,7 +61,7 @@ def test_collect_files_from_dir():
     supported = {".py", ".js", ".ts", ".tsx", ".go", ".rs",
                  ".java", ".c", ".cpp", ".cc", ".cxx", ".rb",
                  ".cs", ".kt", ".kts", ".scala", ".php", ".h", ".hpp",
-                 ".swift", ".lua", ".toc", ".zig", ".ps1"}
+                 ".swift", ".lua", ".toc", ".zig", ".ps1", ".ex", ".exs"}
     assert all(f.suffix in supported for f in files)
     assert len(files) > 0
 

tests/test_languages.py

@@ -337,3 +337,37 @@ def test_swift_emits_calls():
     r = extract_swift(FIXTURES / "sample.swift")
     calls = _calls(r)
     assert any("process" in src and "validate" in tgt for src, tgt in calls)
+
+
+# ── Elixir ────────────────────────────────────────────────────────────────────
+
+from graphify.extract import extract_elixir
+
+def test_elixir_finds_module():
+    r = extract_elixir(FIXTURES / "sample.ex")
+    assert "error" not in r
+    labels = [n["label"] for n in r["nodes"]]
+    assert any("MyApp.Accounts.User" in l for l in labels)
+
+def test_elixir_finds_functions():
+    r = extract_elixir(FIXTURES / "sample.ex")
+    labels = [n["label"] for n in r["nodes"]]
+    assert any("create" in l for l in labels)
+    assert any("find" in l for l in labels)
+    assert any("validate" in l for l in labels)
+
+def test_elixir_finds_imports():
+    r = extract_elixir(FIXTURES / "sample.ex")
+    import_edges = [e for e in r["edges"] if e["relation"] == "imports"]
+    assert len(import_edges) >= 2
+
+def test_elixir_finds_calls():
+    r = extract_elixir(FIXTURES / "sample.ex")
+    calls = {(e["source"], e["target"]) for e in r["edges"] if e["relation"] == "calls"}
+    labels = {n["id"]: n["label"] for n in r["nodes"]}
+    assert any("create" in labels.get(src, "") and "validate" in labels.get(tgt, "") for src, tgt in calls)
+
+def test_elixir_method_edges():
+    r = extract_elixir(FIXTURES / "sample.ex")
+    methods = [e for e in r["edges"] if e["relation"] == "method"]
+    assert len(methods) >= 3