summaryrefslogtreecommitdiffstats
path: root/git_refs.py
diff options
context:
space:
mode:
Diffstat (limited to 'git_refs.py')
-rw-r--r--git_refs.py283
1 files changed, 142 insertions, 141 deletions
diff --git a/git_refs.py b/git_refs.py
index 300d2b30..aca1f90d 100644
--- a/git_refs.py
+++ b/git_refs.py
@@ -16,149 +16,150 @@ import os
16from repo_trace import Trace 16from repo_trace import Trace
17import platform_utils 17import platform_utils
18 18
19HEAD = 'HEAD' 19HEAD = "HEAD"
20R_CHANGES = 'refs/changes/' 20R_CHANGES = "refs/changes/"
21R_HEADS = 'refs/heads/' 21R_HEADS = "refs/heads/"
22R_TAGS = 'refs/tags/' 22R_TAGS = "refs/tags/"
23R_PUB = 'refs/published/' 23R_PUB = "refs/published/"
24R_WORKTREE = 'refs/worktree/' 24R_WORKTREE = "refs/worktree/"
25R_WORKTREE_M = R_WORKTREE + 'm/' 25R_WORKTREE_M = R_WORKTREE + "m/"
26R_M = 'refs/remotes/m/' 26R_M = "refs/remotes/m/"
27 27
28 28
29class GitRefs(object): 29class GitRefs(object):
30 def __init__(self, gitdir): 30 def __init__(self, gitdir):
31 self._gitdir = gitdir 31 self._gitdir = gitdir
32 self._phyref = None 32 self._phyref = None
33 self._symref = None 33 self._symref = None
34 self._mtime = {} 34 self._mtime = {}
35 35
36 @property 36 @property
37 def all(self): 37 def all(self):
38 self._EnsureLoaded() 38 self._EnsureLoaded()
39 return self._phyref 39 return self._phyref
40 40
41 def get(self, name): 41 def get(self, name):
42 try:
43 return self.all[name]
44 except KeyError:
45 return ''
46
47 def deleted(self, name):
48 if self._phyref is not None:
49 if name in self._phyref:
50 del self._phyref[name]
51
52 if name in self._symref:
53 del self._symref[name]
54
55 if name in self._mtime:
56 del self._mtime[name]
57
58 def symref(self, name):
59 try:
60 self._EnsureLoaded()
61 return self._symref[name]
62 except KeyError:
63 return ''
64
65 def _EnsureLoaded(self):
66 if self._phyref is None or self._NeedUpdate():
67 self._LoadAll()
68
69 def _NeedUpdate(self):
70 with Trace(': scan refs %s', self._gitdir):
71 for name, mtime in self._mtime.items():
72 try: 42 try:
73 if mtime != os.path.getmtime(os.path.join(self._gitdir, name)): 43 return self.all[name]
74 return True 44 except KeyError:
45 return ""
46
47 def deleted(self, name):
48 if self._phyref is not None:
49 if name in self._phyref:
50 del self._phyref[name]
51
52 if name in self._symref:
53 del self._symref[name]
54
55 if name in self._mtime:
56 del self._mtime[name]
57
58 def symref(self, name):
59 try:
60 self._EnsureLoaded()
61 return self._symref[name]
62 except KeyError:
63 return ""
64
65 def _EnsureLoaded(self):
66 if self._phyref is None or self._NeedUpdate():
67 self._LoadAll()
68
69 def _NeedUpdate(self):
70 with Trace(": scan refs %s", self._gitdir):
71 for name, mtime in self._mtime.items():
72 try:
73 if mtime != os.path.getmtime(
74 os.path.join(self._gitdir, name)
75 ):
76 return True
77 except OSError:
78 return True
79 return False
80
81 def _LoadAll(self):
82 with Trace(": load refs %s", self._gitdir):
83 self._phyref = {}
84 self._symref = {}
85 self._mtime = {}
86
87 self._ReadPackedRefs()
88 self._ReadLoose("refs/")
89 self._ReadLoose1(os.path.join(self._gitdir, HEAD), HEAD)
90
91 scan = self._symref
92 attempts = 0
93 while scan and attempts < 5:
94 scan_next = {}
95 for name, dest in scan.items():
96 if dest in self._phyref:
97 self._phyref[name] = self._phyref[dest]
98 else:
99 scan_next[name] = dest
100 scan = scan_next
101 attempts += 1
102
103 def _ReadPackedRefs(self):
104 path = os.path.join(self._gitdir, "packed-refs")
105 try:
106 fd = open(path, "r")
107 mtime = os.path.getmtime(path)
108 except IOError:
109 return
75 except OSError: 110 except OSError:
76 return True 111 return
77 return False 112 try:
78 113 for line in fd:
79 def _LoadAll(self): 114 line = str(line)
80 with Trace(': load refs %s', self._gitdir): 115 if line[0] == "#":
81 116 continue
82 self._phyref = {} 117 if line[0] == "^":
83 self._symref = {} 118 continue
84 self._mtime = {} 119
85 120 line = line[:-1]
86 self._ReadPackedRefs() 121 p = line.split(" ")
87 self._ReadLoose('refs/') 122 ref_id = p[0]
88 self._ReadLoose1(os.path.join(self._gitdir, HEAD), HEAD) 123 name = p[1]
89 124
90 scan = self._symref 125 self._phyref[name] = ref_id
91 attempts = 0 126 finally:
92 while scan and attempts < 5: 127 fd.close()
93 scan_next = {} 128 self._mtime["packed-refs"] = mtime
94 for name, dest in scan.items(): 129
95 if dest in self._phyref: 130 def _ReadLoose(self, prefix):
96 self._phyref[name] = self._phyref[dest] 131 base = os.path.join(self._gitdir, prefix)
97 else: 132 for name in platform_utils.listdir(base):
98 scan_next[name] = dest 133 p = os.path.join(base, name)
99 scan = scan_next 134 # We don't implement the full ref validation algorithm, just the
100 attempts += 1 135 # simple rules that would show up in local filesystems.
101 136 # https://git-scm.com/docs/git-check-ref-format
102 def _ReadPackedRefs(self): 137 if name.startswith(".") or name.endswith(".lock"):
103 path = os.path.join(self._gitdir, 'packed-refs') 138 pass
104 try: 139 elif platform_utils.isdir(p):
105 fd = open(path, 'r') 140 self._mtime[prefix] = os.path.getmtime(base)
106 mtime = os.path.getmtime(path) 141 self._ReadLoose(prefix + name + "/")
107 except IOError: 142 else:
108 return 143 self._ReadLoose1(p, prefix + name)
109 except OSError: 144
110 return 145 def _ReadLoose1(self, path, name):
111 try: 146 try:
112 for line in fd: 147 with open(path) as fd:
113 line = str(line) 148 mtime = os.path.getmtime(path)
114 if line[0] == '#': 149 ref_id = fd.readline()
115 continue 150 except (OSError, UnicodeError):
116 if line[0] == '^': 151 return
117 continue 152
118 153 try:
119 line = line[:-1] 154 ref_id = ref_id.decode()
120 p = line.split(' ') 155 except AttributeError:
121 ref_id = p[0] 156 pass
122 name = p[1] 157 if not ref_id:
123 158 return
124 self._phyref[name] = ref_id 159 ref_id = ref_id[:-1]
125 finally: 160
126 fd.close() 161 if ref_id.startswith("ref: "):
127 self._mtime['packed-refs'] = mtime 162 self._symref[name] = ref_id[5:]
128 163 else:
129 def _ReadLoose(self, prefix): 164 self._phyref[name] = ref_id
130 base = os.path.join(self._gitdir, prefix) 165 self._mtime[name] = mtime
131 for name in platform_utils.listdir(base):
132 p = os.path.join(base, name)
133 # We don't implement the full ref validation algorithm, just the simple
134 # rules that would show up in local filesystems.
135 # https://git-scm.com/docs/git-check-ref-format
136 if name.startswith('.') or name.endswith('.lock'):
137 pass
138 elif platform_utils.isdir(p):
139 self._mtime[prefix] = os.path.getmtime(base)
140 self._ReadLoose(prefix + name + '/')
141 else:
142 self._ReadLoose1(p, prefix + name)
143
144 def _ReadLoose1(self, path, name):
145 try:
146 with open(path) as fd:
147 mtime = os.path.getmtime(path)
148 ref_id = fd.readline()
149 except (OSError, UnicodeError):
150 return
151
152 try:
153 ref_id = ref_id.decode()
154 except AttributeError:
155 pass
156 if not ref_id:
157 return
158 ref_id = ref_id[:-1]
159
160 if ref_id.startswith('ref: '):
161 self._symref[name] = ref_id[5:]
162 else:
163 self._phyref[name] = ref_id
164 self._mtime[name] = mtime
generated by cgit v1.2.3-54-g00ecf (git 2.39.0) at 2025-10-25 21:28:47 +0000