1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
"""A class for remote git repos"""
from os import path
from shutil import rmtree
from subprocess import call
from tempfile import mkdtemp
from git import Repo
from pomu.repo.remote.remote import RemoteRepo, normalize_key
from pomu.util.git import parse_object
from pomu.util.result import Result
class RemoteGitRepo(RemoteRepo):
"""A class responsible for git remotes"""
def __init__(self, url):
self.uri = url
self.dir = mkdtemp()
if call(['git', 'clone', '--depth=1', '--bare', url, self.dir]) > 0: # we've a problem
raise RuntimeError()
self.repo = Repo(self.dir)
def __enter__(self):
pass
def __exit__(self, *_):
self.cleanup()
def get_object(self, oid):
head, tail = oid[0:2], oid[2:]
opath = path.join(self.dir, 'objects', head, tail)
return open(opath, 'rb').read()
def _fetch_tree(self, obj, tpath):
res = []
ents = parse_object(self.get_object(obj), tpath).unwrap()
for is_dir, sha, opath in ents:
res.append((opath.decode('utf-8') + ('/' if is_dir else ''), sha))
if is_dir:
res.extend(self._fetch_tree(sha, opath))
return res
def fetch_tree(self):
"""Returns repos hierarchy"""
if hasattr(self, '_tree'):
return [x for x, y in self._tree]
tid = self.repo.tree().hexsha
res = self._fetch_tree(tid, b'')
self._tree = res
return [x for x, y in res]
def fetch_subtree(self, key):
"""Lists a subtree"""
k = normalize_key(key, True)
self.fetch_tree()
dic = dict(self._tree)
if k not in dic:
return Result.Err()
l = len(key)
return Result.Ok(
[tpath[l:] for tpath in self.fetch_tree() if tpath.startswith(k)])
def fetch_file(self, key):
"""Fetches a file from the repo"""
k = normalize_key(key)
self.fetch_tree()
dic = dict(self._tree)
if k not in dic:
return Result.Err()
return parse_object(self.get_object(dic[k]))
def cleanup(self):
rmtree(self.dir)
|
GitWeb
