sqlmesh.utils.git

 1from __future__ import annotations
 2
 3import subprocess
 4import typing as t
 5from functools import cached_property
 6from pathlib import Path
 7
 8
 9class GitClient:
10    def __init__(self, repo: str | Path):
11        self._work_dir = Path(repo)
12
13    def list_untracked_files(self) -> t.List[Path]:
14        return self._execute_list_output(
15            ["ls-files", "--others", "--exclude-standard"], self._work_dir
16        )
17
18    def list_uncommitted_changed_files(self) -> t.List[Path]:
19        return self._execute_list_output(
20            ["diff", "--name-only", "--diff-filter=d", "HEAD"], self._git_root
21        )
22
23    def list_committed_changed_files(self, target_branch: str = "main") -> t.List[Path]:
24        return self._execute_list_output(
25            ["diff", "--name-only", "--diff-filter=d", f"{target_branch}..."], self._git_root
26        )
27
28    def _execute_list_output(self, commands: t.List[str], base_path: Path) -> t.List[Path]:
29        return [(base_path / o).absolute() for o in self._execute(commands).split("\n") if o]
30
31    def _execute(self, commands: t.List[str]) -> str:
32        result = subprocess.run(
33            ["git"] + commands,
34            cwd=self._work_dir,
35            stdout=subprocess.PIPE,
36            stderr=subprocess.PIPE,
37            check=False,
38        )
39
40        # If the Git command failed, extract and raise the error message in the console
41        if result.returncode != 0:
42            stderr_output = result.stderr.decode("utf-8").strip()
43            error_message = next(
44                (line for line in stderr_output.splitlines() if line.lower().startswith("fatal:")),
45                stderr_output,
46            )
47            raise RuntimeError(f"Git error: {error_message}")
48
49        return result.stdout.decode("utf-8").strip()
50
51    @cached_property
52    def _git_root(self) -> Path:
53        return Path(self._execute(["rev-parse", "--show-toplevel"]))