Coverage for tarjinja/git.py: 30%
36 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-05 14:24 +0000
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-05 14:24 +0000
1import io
2import os
3import tempfile
4import time
5from collections.abc import Generator
6from logging import getLogger
8import git
10from .iface import Input
12log = getLogger(__name__)
15class GitInput(Input):
16 def __init__(
17 self, ifn: str, name: str, branch: str = "master", tmpdir: str | None = None
18 ):
19 log.debug("ifn=%s, name=%s, branch=%s, tmpdir=%s", ifn, name, branch, tmpdir)
20 super().__init__(ifn)
21 if tmpdir is None:
22 self.tmpd = tempfile.TemporaryDirectory()
23 self.tmpdir = self.tmpd.name
24 else:
25 self.tmpdir = tmpdir
26 dirn = os.path.join(self.tmpdir, name)
27 if os.path.isdir(dirn):
28 self.repo = git.repo.Repo(dirn)
29 self.repo.remote().fetch(refspec=branch, depth=1)
30 else:
31 self.repo = git.repo.Repo.clone_from(
32 self.ifn, dirn, bare=True, branch=branch, depth=1
33 )
34 self.ts = time.time()
35 self.files = {}
37 def walk(self) -> Generator[tuple[str, int, float], None, None]:
38 for x in self.repo.tree().traverse():
39 if x.type != "blob":
40 continue
41 self.files[x.path] = x
42 yield x.path, x.mode, self.ts
44 def readfile(self, fn: str) -> str:
45 try:
46 return io.TextIOWrapper(
47 io.BytesIO(self.files.get(fn).data_stream.read())
48 ).read()
49 except UnicodeDecodeError as e:
50 log.warning("cannot decode: %s", e)
51 return ""