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(["diff", "--name-only", "--diff-filter=d"], self._git_root)
20
21    def list_committed_changed_files(self, target_branch: str = "main") -> t.List[Path]:
22        return self._execute_list_output(
23            ["diff", "--name-only", "--diff-filter=d", f"{target_branch}..."], self._git_root
24        )
25
26    def _execute_list_output(self, commands: t.List[str], base_path: Path) -> t.List[Path]:
27        return [(base_path / o).absolute() for o in self._execute(commands).split("\n") if o]
28
29    def _execute(self, commands: t.List[str]) -> str:
30        result = subprocess.run(["git"] + commands, cwd=self._work_dir, stdout=subprocess.PIPE)
31        return result.stdout.decode("utf-8").strip()
32
33    @cached_property
34    def _git_root(self) -> Path:
35        return Path(self._execute(["rev-parse", "--show-toplevel"]))