From 3d5da6039a59b237ab401295cda0817a76e6429f Mon Sep 17 00:00:00 2001 From: Safi Date: Tue, 7 Apr 2026 12:54:12 +0100 Subject: [PATCH] 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. --- graphify/detect.py | 2 +- graphify/extract.py | 179 +++++++++++++++++++++++++++++++++++++++ pyproject.toml | 3 +- tests/fixtures/sample.ex | 28 ++++++ tests/test_extract.py | 2 +- tests/test_languages.py | 34 ++++++++ 6 files changed, 245 insertions(+), 3 deletions(-) create mode 100644 tests/fixtures/sample.ex diff --git a/graphify/detect.py b/graphify/detect.py index 6d1cae08..96b1d57a 100644 --- a/graphify/detect.py +++ b/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'} diff --git a/graphify/extract.py b/graphify/extract.py index 3191e424..217caec5 100644 --- a/graphify/extract.py +++ b/graphify/extract.py @@ -1928,6 +1928,183 @@ def _resolve_cross_file_imports( 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: diff --git a/pyproject.toml b/pyproject.toml index 82e574a6..4ae7283b 100644 --- a/pyproject.toml +++ b/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] diff --git a/tests/fixtures/sample.ex b/tests/fixtures/sample.ex new file mode 100644 index 00000000..9e048216 --- /dev/null +++ b/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 diff --git a/tests/test_extract.py b/tests/test_extract.py index e95b273f..930e4080 100644 --- a/tests/test_extract.py +++ b/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 diff --git a/tests/test_languages.py b/tests/test_languages.py index 7f645b85..d4598533 100644 --- a/tests/test_languages.py +++ b/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