Something went wrong. Try again.
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006"""Git repos as Python objects: reads straight from .git, git itself for the rest.
from pyrepl import git git.status() branch, upstream, ahead/behind, files by state git.log(10) commits, newest first; rev=, path=, first_parent= git.show("HEAD~1") one commit and the files it changed git.diff("a.py", staged=True) diff text; rev="main" diffs against a revision git.changes(rev=None) [(status, path)] as `git diff --name-status` git.cat("HEAD:src/x.py") a file's bytes at a revision; git.ls("HEAD", "src") git.blame("x.py", 10, 20) lines with the commit that last touched each git.get("v1.2"), git.head(), git.branch(), git.branches(), git.tags(), git.resolve("main~2"), git.obj(sha) -> (type, bytes) git.root(), git.config("user.email") git.add("a.py"), git.commit("message", all=False) -> the new Commit git("log --oneline -3") anything else: git with those args, no shell git.at("~/src/x") all of the above in another repo
What can be read is read natively: HEAD, loose and packed refs, loose andpacked objects (deltas included), commits, trees, tags, worktrees andalternates; log() walks commits itself and show() diffs trees itself.status, diff, blame, config and every write go through git (the index,hooks and locks are git's business), with -z machine formats and argv,never a shell. When the native side meets something it doesn't do (aSHA-256 or reftable repo, an old pack index), that call falls back to git.
Commits come back as Commit objects (.sha .parents .author .date .message.summary ...), and everything prints as git would show it.
Must stay runnable on python 3.6 (sessions can run on any 3.6+)."""import binasciiimport datetimeimport heapqimport osimport reimport shleximport subprocessimport sysimport typesimport zlib
try: from tk import Out, Lines, CommandError, _resultexcept ImportError: # a session started with PYREPL_NO_TK class Out(str): def __pyrepl_show__(self): return str(self)
class Lines(list): def __pyrepl_show__(self): return "\n".join(self) or "(none)"
class CommandError(RuntimeError): def __init__(self, result): self.result = result RuntimeError.__init__(self, "exit %s: %s\n%s" % (result.code, result.cmd, result.err))
def _result(text, code, cmd, err=""): r = Out(text) r.code, r.cmd, r.err, r.ok = code, cmd, err, code == 0 return r
__all__ = ["Repo", "Commit", "GitError", "at", "repo", "run", "status", "log", "show", "diff", "changes", "cat", "ls", "blame", "head", "branch", "branches", "tags", "resolve", "root", "config", "add", "commit", "get", "refs", "obj"]
HEX40 = re.compile(r"^[0-9a-f]{40}$")HASH = re.compile(r"^[0-9a-f]{40}(?:[0-9a-f]{24})?$") # SHA-1 or SHA-256HEXPART = re.compile(r"^[0-9a-f]{4,40}$")TYPES = {1: "commit", 2: "tree", 3: "blob", 4: "tag"}
class GitError(RuntimeError): pass
class _Fallback(Exception): """The native side can't do this one: ask git."""
# -- running git ----------------------------------------------------------------def _git(cwd, args, input=None, check=True, timeout=120, text=True): """git with these args in cwd: argv, never a shell. text=False: bytes out.""" argv = ["git"] + [str(a) for a in args] env = dict(os.environ, GIT_PAGER="cat", PAGER="cat", GIT_TERMINAL_PROMPT="0", GIT_OPTIONAL_LOCKS="0") if isinstance(input, str): input = input.encode("utf-8") p = subprocess.run(argv, cwd=cwd, env=env, input=input, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout) out = p.stdout.decode("utf-8", "replace") if text else p.stdout r = _result(out if text else "", p.returncode, " ".join(shlex.quote(a) for a in argv), p.stderr.decode("utf-8", "replace")) if check and p.returncode: raise CommandError(r) return out if not text else r
# -- what comes back --------------------------------------------------------------def _when(stamp): """"1700000000 +0100" -> aware datetime.""" secs, _, tz = stamp.partition(" ") sign = -1 if tz.startswith("-") else 1 off = datetime.timedelta(hours=int(tz[1:3] or 0), minutes=int(tz[3:5] or 0)) * sign return datetime.datetime.fromtimestamp(int(secs), datetime.timezone(off))
def _person(value): """"Name <email> 1700000000 +0100" -> (name, email, datetime).""" m = re.match(r"^(.*?) <(.*?)> (\d+ [+-]\d{4})$", value) if not m: return value, "", None return m.group(1), m.group(2), _when(m.group(3))
class Commit(object): """A commit, parsed from its object."""
def __init__(self, sha, raw, refs=()): self.sha = sha self.parents = [] self.tree = self.author = self.author_email = self.author_date = None self.committer = self.committer_email = self.date = None self.headers = {} head, _, msg = raw.partition(b"\n\n") key = None for line in head.decode("utf-8", "replace").split("\n"): if line.startswith(" ") and key: # continuation (gpgsig, mergetag) self.headers[key] += "\n" + line[1:] continue key, _, value = line.partition(" ") if key == "parent": self.parents.append(value) elif key == "tree": self.tree = value elif key == "author": self.author, self.author_email, self.author_date = _person(value) elif key == "committer": self.committer, self.committer_email, self.date = _person(value) else: self.headers[key] = value self.message = msg.decode(self.headers.get("encoding", "utf-8"), "replace") self.refs = list(refs) self.files = None # show() fills these in: [(status, path)]
@property def summary(self): return self.message.split("\n", 1)[0]
@property def short(self): return self.sha[:9]
def oneline(self): refs = " (%s)" % ", ".join(self.refs) if self.refs else "" when = self.author_date.strftime("%Y-%m-%d") if self.author_date else "?" return "%s %s %-16.16s %s%s" % (self.short, when, self.author or "?", self.summary, refs)
def __pyrepl_show__(self): out = ["commit %s%s" % (self.sha, " (%s)" % ", ".join(self.refs) if self.refs else "")] if len(self.parents) > 1: out.append("Merge: " + " ".join(p[:9] for p in self.parents)) out.append("Author: %s <%s>" % (self.author, self.author_email)) out.append("Date: %s" % (self.author_date.strftime("%a %Y-%m-%d %H:%M:%S %z") if self.author_date else "?")) out.append("") out.extend(" " + l for l in self.message.rstrip("\n").split("\n")) if self.files is not None: out.append("") out.extend(" %s %s" % (s, p) for s, p in self.files) if not self.files: out.append(" (no file changes)") return "\n".join(out)
def __repr__(self): return "<Commit %s %s>" % (self.short, self.summary[:60])
class Log(list): """Commits, newest first."""
def __pyrepl_show__(self): return "\n".join(c.oneline() for c in self) or "(no commits)"
class Status(object): """`git status --porcelain=v2`, parsed. Paths are relative to the root."""
def __init__(self): self.oid = self.branch = self.upstream = None self.ahead = self.behind = 0 self.staged, self.unstaged, self.untracked, self.conflicts = [], [], [], []
@property def clean(self): return not (self.staged or self.unstaged or self.untracked or self.conflicts)
def __pyrepl_show__(self): head = "## %s" % (self.branch or "HEAD (no branch)") if self.upstream: head += "...%s" % self.upstream ab = ", ".join(x for x in ("ahead %d" % self.ahead if self.ahead else "", "behind %d" % self.behind if self.behind else "") if x) head += " [%s]" % ab if ab else "" out = [head] out += ["UU %s" % p for p in self.conflicts] out += ["%s %s" % (s, p) for s, p in self.staged] out += [" %s %s" % (s, p) for s, p in self.unstaged] out += ["?? %s" % p for p in self.untracked] if self.clean: out.append("(clean)") return "\n".join(out)
def __repr__(self): return "<Status %s: %d staged, %d unstaged, %d untracked%s>" % ( self.branch, len(self.staged), len(self.unstaged), len(self.untracked), ", %d conflicts" % len(self.conflicts) if self.conflicts else "")
class Refs(dict): """name -> sha; `current` is the checked-out branch, if any."""
current = None
def __pyrepl_show__(self): if not self: return "(none)" return "\n".join("%s %s %s" % ("*" if n == self.current else " ", s[:9], n) for n, s in sorted(self.items()))
class Entry(tuple): """A tree entry: (mode, type, sha, name)."""
mode, type, sha, name = (property(lambda s, i=i: s[i]) for i in range(4))
class Tree(list): def __pyrepl_show__(self): return "\n".join("%s %s %s\t%s" % e for e in self) or "(empty)"
class BlameLine(tuple): """(line number, sha, author, datetime, text)"""
lineno, sha, author, date, text = (property(lambda s, i=i: s[i]) for i in range(5))
class Blame(list): def __pyrepl_show__(self): return "\n".join("%s %-12.12s %s %4d) %s" % ( b.sha[:9], b.author, b.date.strftime("%Y-%m-%d") if b.date else "?", b.lineno, b.text) for b in self) or "(empty)"
# -- packs ----------------------------------------------------------------------------class _Pack(object): """One .pack and its v2 .idx, read on first use."""
def __init__(self, idx_path): self.idx_path = idx_path self.pack_path = idx_path[:-4] + ".pack" with open(idx_path, "rb") as fh: data = fh.read() if data[:4] != b"\xfftOc" or int.from_bytes(data[4:8], "big") != 2: raise _Fallback("pack index %s is not version 2" % idx_path) self.data = data self.fanout = [int.from_bytes(data[8 + 4 * i:12 + 4 * i], "big") for i in range(256)] self.n = self.fanout[255] self.shas = 8 + 1024 self.offsets = self.shas + 24 * self.n # past the sha and crc tables self.large = self.offsets + 4 * self.n
def _sha_at(self, i): return self.data[self.shas + 20 * i:self.shas + 20 * i + 20]
def find(self, binsha): lo = self.fanout[binsha[0] - 1] if binsha[0] else 0 hi = self.fanout[binsha[0]] while lo < hi: mid = (lo + hi) // 2 s = self._sha_at(mid) if s < binsha: lo = mid + 1 elif s > binsha: hi = mid else: return self._offset(mid) return None
def prefixed(self, binpre, odd_nibble=None): """Full hex shas starting with this prefix (for short shas).""" lo = self.fanout[binpre[0] - 1] if binpre[0] else 0 hi = self.fanout[binpre[0]] out = [] for i in range(lo, hi): s = self._sha_at(i) if s.startswith(binpre) and (odd_nibble is None or s[len(binpre)] >> 4 == odd_nibble): out.append(binascii.hexlify(s).decode()) return out
def _offset(self, i): off = int.from_bytes(self.data[self.offsets + 4 * i:self.offsets + 4 * i + 4], "big") if off & 0x80000000: j = off & 0x7fffffff off = int.from_bytes(self.data[self.large + 8 * j:self.large + 8 * j + 8], "big") return off
def _varint_header(fh): """Type and size from a pack entry header.""" c = fh.read(1)[0] typ, size, shift = (c >> 4) & 7, c & 15, 4 while c & 0x80: c = fh.read(1)[0] size |= (c & 0x7f) << shift shift += 7 return typ, size
def _inflate(fh): d = zlib.decompressobj() out = [] while not d.eof: chunk = fh.read(65536) if not chunk: raise _Fallback("truncated pack entry") out.append(d.decompress(chunk)) return b"".join(out)
def _apply_delta(base, delta): def varint(pos): n = shift = 0 while True: c = delta[pos] pos += 1 n |= (c & 0x7f) << shift shift += 7 if not c & 0x80: return n, pos src, pos = varint(0) dst, pos = varint(pos) if src != len(base): raise _Fallback("delta base size mismatch") out = bytearray() while pos < len(delta): op = delta[pos] pos += 1 if op & 0x80: # copy from base off = size = 0 for i in range(4): if op & (1 << i): off |= delta[pos] << (8 * i) pos += 1 for i in range(3): if op & (1 << (4 + i)): size |= delta[pos] << (8 * i) pos += 1 out += base[off:off + (size or 0x10000)] elif op: # insert literal out += delta[pos:pos + op] pos += op else: raise _Fallback("bad delta opcode") if len(out) != dst: raise _Fallback("delta result size mismatch") return bytes(out)
# -- a repository --------------------------------------------------------------------class Repo(object): """A git repository: the worktree it was found from, its git dir, and its common dir (where objects and refs live; the same unless this is a linked worktree)."""
def __init__(self, path="."): path = os.path.abspath(os.path.expanduser(str(path))) d = path if os.path.isdir(path) else os.path.dirname(path) while True: dotgit = os.path.join(d, ".git") if os.path.isdir(dotgit): self.root, self.gitdir = d, dotgit break if os.path.isfile(dotgit): # worktree or submodule: "gitdir: <path>" with open(dotgit) as fh: target = fh.read().strip() if not target.startswith("gitdir: "): raise GitError("%s is not a gitdir link" % dotgit) self.root = d self.gitdir = os.path.normpath(os.path.join(d, target[8:])) break if (os.path.isfile(os.path.join(d, "HEAD")) and os.path.isdir(os.path.join(d, "objects")) and os.path.isdir(os.path.join(d, "refs"))): self.root, self.gitdir = None, d # bare break up = os.path.dirname(d) if up == d: raise GitError("not in a git repository: %s" % path) d = up self.common = self.gitdir cd = os.path.join(self.gitdir, "commondir") if os.path.isfile(cd): with open(cd) as fh: self.common = os.path.normpath(os.path.join(self.gitdir, fh.read().strip())) self.cwd = self.root or self.gitdir self.native, self.hashlen = True, 20 cfg = os.path.join(self.common, "config") try: with open(cfg, encoding="utf-8", errors="replace") as fh: local = fh.read().lower() if re.search(r"objectformat\s*=\s*sha256", local): self.hashlen = 32 if self.hashlen != 20 or re.search(r"refstorage\s*=\s*reftable", local): self.native = False # git reads these; we don't except OSError: pass self._packs, self._packs_key, self._cache = [], None, {}
def __repr__(self): return "<git repo %s>" % (self.root or self.gitdir)
# -- git itself def run(self, *args, **kw): """git with these args in this repo, no shell. One string is split like a shell would (no expansion). Raises CommandError on failure unless check=False; input= feeds stdin.""" if len(args) == 1 and isinstance(args[0], str): args = shlex.split(args[0]) return _git(self.cwd, args, kw.get("input"), kw.get("check", True), kw.get("timeout", 120))
__call__ = run
# -- refs def _packed_refs(self): refs, peeled, last = {}, {}, None try: with open(os.path.join(self.common, "packed-refs"), encoding="utf-8") as fh: for line in fh: line = line.rstrip("\n") if line.startswith("#") or not line: continue if line.startswith("^"): if last: peeled[last] = line[1:] continue sha, _, name = line.partition(" ") refs[name], last = sha, name except FileNotFoundError: pass return refs, peeled
def _ref_file(self, name): # HEAD and per-worktree refs live in the worktree's gitdir, the rest in common. per_tree = name == "HEAD" or name.startswith(("refs/bisect/", "refs/worktree/", "refs/rewritten/")) or "/" not in name p = os.path.join(self.gitdir if per_tree else self.common, name) try: with open(p, encoding="utf-8") as fh: return fh.read().strip() except (FileNotFoundError, NotADirectoryError, IsADirectoryError): return None
def ref(self, name, depth=0): """sha a ref points to (following symbolic refs), or None.""" if not self.native: r = self.run("rev-parse", "-q", "--verify", name, check=False) return r.strip() or None v = self._ref_file(name) if v is None: v = self._packed_refs()[0].get(name) if v is None: return None if v.startswith("ref: "): return self.ref(v[5:], depth + 1) if depth < 10 else None return v
def head(self): """{"branch": name or None (detached), "sha": sha or None (no commits yet)}.""" v = self._ref_file("HEAD") if self.native else None if v is None: b = self.run("symbolic-ref", "-q", "--short", "HEAD", check=False).strip() return {"branch": b or None, "sha": self.ref("HEAD")} if v.startswith("ref: "): name = v[5:] return {"branch": name[11:] if name.startswith("refs/heads/") else name, "sha": self.ref(name)} return {"branch": None, "sha": v}
def branch(self): return self.head()["branch"]
def refs(self, prefix="refs/"): """{full ref name: sha} for loose and packed refs under prefix.""" if not self.native: out = self.run("for-each-ref", "--format=%(objectname) %(refname)", prefix) return dict(reversed(l.split(" ", 1)) for l in out.splitlines()) refs = dict((n, s) for n, s in self._packed_refs()[0].items() if n.startswith(prefix)) top = os.path.join(self.common, prefix.rstrip("/")) for dp, _, fns in os.walk(top): for fn in fns: p = os.path.join(dp, fn) name = os.path.relpath(p, self.common).replace(os.sep, "/") if not name.startswith(prefix): continue try: with open(p, encoding="utf-8") as fh: v = fh.read().strip() except OSError: continue if HEX40.match(v): refs[name] = v elif v.startswith("ref: "): sha = self.ref(v[5:]) if sha: refs[name] = sha return refs
def branches(self, remote=False): prefix = "refs/remotes/" if remote else "refs/heads/" out = Refs((n[len(prefix):], s) for n, s in self.refs(prefix).items() if not n.endswith("/HEAD")) out.current = self.branch() return out
def tags(self): return Refs((n[10:], s) for n, s in self.refs("refs/tags/").items())
def _decorations(self): dec = {} h = self.head() if h["sha"]: dec.setdefault(h["sha"], []).append("HEAD -> " + h["branch"] if h["branch"] else "HEAD") for n, s in self.refs().items(): short = re.sub(r"^refs/(heads|remotes)/", "", n) short = re.sub(r"^refs/tags/", "tag: ", short) if n.startswith("refs/heads/") and h["branch"] == short: continue if n.startswith("refs/tags/"): s = self._peel(s) dec.setdefault(s, []).append(short) return dec
# -- objects def _pack_list(self): pdir = os.path.join(self.common, "objects", "pack") try: key = os.stat(pdir).st_mtime_ns except FileNotFoundError: return [] if key != self._packs_key: self._packs = [_Pack(p) for p in sorted( os.path.join(pdir, f) for f in os.listdir(pdir) if f.endswith(".idx"))] self._packs_key = key return self._packs
def _object_dirs(self): dirs = [os.path.join(self.common, "objects")] alt = os.path.join(dirs[0], "info", "alternates") if os.path.isfile(alt): with open(alt) as fh: for line in fh: line = line.strip() if line and not line.startswith("#"): dirs.append(os.path.normpath(os.path.join(dirs[0], line))) return dirs
def _native_object(self, sha): hit = self._cache.get(sha) if hit: return hit for odir in self._object_dirs(): p = os.path.join(odir, sha[:2], sha[2:]) if os.path.isfile(p): with open(p, "rb") as fh: raw = zlib.decompress(fh.read()) head, _, body = raw.partition(b"\0") return self._remember(sha, (head.split(b" ")[0].decode(), body)) binsha = binascii.unhexlify(sha) packs = self._pack_list() for odir in self._object_dirs()[1:]: # an alternate's packs pdir = os.path.join(odir, "pack") if os.path.isdir(pdir): packs = packs + [_Pack(os.path.join(pdir, f)) for f in sorted(os.listdir(pdir)) if f.endswith(".idx")] for pack in packs: off = pack.find(binsha) if off is not None: return self._remember(sha, self._from_pack(pack, off)) raise KeyError(sha)
def _remember(self, sha, obj): if len(self._cache) > 512: self._cache.clear() self._cache[sha] = obj return obj
def _from_pack(self, pack, off): """Follow a delta chain down to its base, then apply it back up.""" deltas = [] with open(pack.pack_path, "rb") as fh: while True: fh.seek(off) typ, _ = _varint_header(fh) if typ == 6: # offset delta c = fh.read(1)[0] rel = c & 0x7f while c & 0x80: c = fh.read(1)[0] rel = ((rel + 1) << 7) | (c & 0x7f) deltas.append(_inflate(fh)) off -= rel elif typ == 7: # ref delta: base by sha base = binascii.hexlify(fh.read(20)).decode() deltas.append(_inflate(fh)) typ, data = self.obj(base) break elif typ in TYPES: typ, data = TYPES[typ], _inflate(fh) break else: raise _Fallback("pack entry type %d" % typ) for d in reversed(deltas): data = _apply_delta(data, d) return typ, data
def obj(self, sha): """(type, bytes) of an object by full sha: natively, else from git.""" if self.native: try: return self._native_object(sha) except (_Fallback, KeyError, OSError, zlib.error, IndexError, ValueError): pass typ = self.run("cat-file", "-t", sha).strip() return typ, _git(self.cwd, ["cat-file", typ, sha], text=False)
def _peel(self, sha): """A tag's target, all the way down.""" for _ in range(10): typ, data = self.obj(sha) if typ != "tag": return sha sha = data.split(b"\n", 1)[0].split(b" ")[1].decode() return sha
def get(self, rev="HEAD"): """The Commit a revision names.""" sha = self.resolve(rev) typ, data = self.obj(sha) if typ != "commit": raise GitError("%s is a %s, not a commit" % (rev, typ)) return Commit(sha, data, self._decorations().get(sha, ()))
# -- revisions def resolve(self, rev="HEAD"): """Full sha of a commit-ish: sha (short too), ref or branch/tag name, with ~N / ^N suffixes. Anything fancier (@{u}, :/text, dates) goes through git rev-parse.""" if self.native: try: return self._resolve_native(rev) except _Fallback: pass r = self.run("rev-parse", "-q", "--verify", "--end-of-options", rev + "^{commit}", check=False) if not r.ok or not r.strip(): raise GitError("unknown revision %r" % rev) return r.strip()
def _resolve_native(self, rev): m = re.match(r"^(.*?)((?:[~^]\d*)*)$", rev) base, ops = m.group(1), m.group(2) if not base or re.search(r"[@{}:\\]|\.\.|^[/.-]", base): raise _Fallback() # git's syntax, or not a ref name at all sha = base if HEX40.match(base) else None # git's order: $GIT_DIR/<name> (HEAD and friends), refs/<name>, tags, # branches, remotes, a remote's HEAD; then a short sha. names = [base] if base.isupper() or base.startswith("refs/") else [] names += ["refs/" + base, "refs/tags/" + base, "refs/heads/" + base, "refs/remotes/" + base, "refs/remotes/%s/HEAD" % base] for name in names if not sha else (): sha = self.ref(name) if sha: break if not sha and HEXPART.match(base): sha = self._unique_prefix(base) if not sha: raise _Fallback() sha = self._peel(sha) for op, n in re.findall(r"([~^])(\d*)", ops): n = int(n) if n else 1 if op == "~": for _ in range(n): sha = self._parents(sha)[0] elif n: sha = self._parents(sha)[n - 1] return sha
def _parents(self, sha): typ, data = self.obj(sha) ps = [l[7:].decode() for l in data.split(b"\n\n", 1)[0].split(b"\n") if l.startswith(b"parent ")] if not ps: raise GitError("%s has no parent" % sha[:9]) return ps
def _unique_prefix(self, pre): found = set() for odir in self._object_dirs(): d = os.path.join(odir, pre[:2]) if os.path.isdir(d) and len(pre) >= 2: found.update(pre[:2] + f for f in os.listdir(d) if f.startswith(pre[2:])) even = pre[:len(pre) // 2 * 2] odd = int(pre[-1], 16) if len(pre) % 2 else None for pack in self._pack_list(): found.update(pack.prefixed(binascii.unhexlify(even), odd)) if len(found) == 1: return found.pop() if len(found) > 1: raise GitError("short sha %s is ambiguous" % pre) return None
# -- trees and files def _tree(self, sha): typ, data = self.obj(sha) if typ == "commit": return self._tree(self._tree_sha(sha)) if typ != "tree": raise GitError("%s is a %s, not a tree" % (sha[:9], typ)) out, i = [], 0 while i < len(data): sp = data.index(b" ", i) nul = data.index(b"\0", sp) mode = data[i:sp].decode() name = data[sp + 1:nul].decode("utf-8", "surrogateescape") esha = binascii.hexlify(data[nul + 1:nul + 1 + self.hashlen]).decode() kind = "tree" if mode == "40000" else "commit" if mode == "160000" else "blob" out.append(Entry((mode.rjust(6, "0"), kind, esha, name))) i = nul + 1 + self.hashlen return out
def _walk_to(self, rev, path): """The tree entry at path in rev (None for the root tree's sha).""" sha = self._tree_sha(self.resolve(rev)) parts = [p for p in path.strip("/").split("/") if p] entry = None for i, part in enumerate(parts): for e in self._tree(sha): if e.name == part: entry, sha = e, e.sha break else: raise GitError("%s does not exist in %s" % ("/".join(parts[:i + 1]), rev)) return entry, sha
def _tree_sha(self, commit_sha): typ, data = self.obj(commit_sha) return data.split(b"\n", 1)[0][5:].decode()
def ls(self, rev="HEAD", path=""): """The entries of a directory at a revision.""" entry, sha = self._walk_to(rev, path) if entry is not None and entry.type != "tree": return Tree([entry]) return Tree(self._tree(sha))
def cat(self, spec, path=None): """A file's bytes at a revision: cat("HEAD:a/b.py") or cat("HEAD", "a/b.py").""" if path is None: spec, _, path = spec.partition(":") spec = spec or "HEAD" entry, sha = self._walk_to(spec, path) if entry is None or entry.type != "blob": raise GitError("%s:%s is not a file" % (spec, path)) return self.obj(sha)[1]
def _tree_changes(self, old, new, prefix=""): """[(A|M|D|T, path)] between two tree shas, recursing only where they differ.""" if old == new: return [] a = dict((e.name, e) for e in self._tree(old)) if old else {} b = dict((e.name, e) for e in self._tree(new)) if new else {} out = [] for name in sorted(set(a) | set(b)): ea, eb = a.get(name), b.get(name) p = prefix + name if ea and eb and ea.sha == eb.sha and ea.mode == eb.mode: continue ta = ea and ea.type == "tree" tb = eb and eb.type == "tree" if ta or tb: out += self._tree_changes(ea.sha if ta else None, eb.sha if tb else None, p + "/") if ea and not ta: out.append(("D", p)) if eb and not tb: out.append(("A", p)) elif not ea: out.append(("A", p)) elif not eb: out.append(("D", p)) else: out.append(("T" if ea.mode != eb.mode and ea.sha == eb.sha else "M", p)) return out
def show(self, rev="HEAD"): """A commit with the files it changed against its first parent (renames show as D + A; git.diff(rev=...) for the patch).""" c = self.get(rev) base = self._tree_sha(c.parents[0]) if c.parents else None c.files = self._tree_changes(base, c.tree) return c
# -- history def log(self, n=15, rev="HEAD", path=None, first_parent=False): """Up to n commits reachable from rev, newest (by commit date) first. Walked natively; with path, git picks the commits that touch it.""" dec = self._decorations() if path is not None: args = ["log", "-z", "--format=%H", "-n", n] + (["--first-parent"] if first_parent else []) + [rev, "--", path] shas = [s for s in self.run(*args).split("\0") if s.strip()] return Log(Commit(s.strip(), self.obj(s.strip())[1], dec.get(s.strip(), ())) for s in shas) try: start = self.resolve(rev) except GitError: if rev == "HEAD" and not self.head()["sha"]: return Log() # no commits yet raise out, seen, heap, k = Log(), {start}, [], 0 c = Commit(start, self.obj(start)[1], dec.get(start, ())) heapq.heappush(heap, (-(c.date.timestamp() if c.date else 0), k, c)) while heap and len(out) < n: _, _, c = heapq.heappop(heap) out.append(c) for p in c.parents[:1] if first_parent else c.parents: if p not in seen: seen.add(p) k += 1 pc = Commit(p, self.obj(p)[1], dec.get(p, ())) heapq.heappush(heap, (-(pc.date.timestamp() if pc.date else 0), k, pc)) return out
# -- the working tree, through git def status(self): raw = _git(self.cwd, ["status", "--porcelain=v2", "-z", "--branch", "--untracked-files=normal"], text=False) st = Status() items = raw.decode("utf-8", "surrogateescape").split("\0") i = 0 while i < len(items): it = items[i] i += 1 if it.startswith("# branch.oid "): st.oid = it[13:] if it[13:] != "(initial)" else None elif it.startswith("# branch.head "): st.branch = it[14:] if it[14:] != "(detached)" else None elif it.startswith("# branch.upstream "): st.upstream = it[18:] elif it.startswith("# branch.ab "): a, b = it[12:].split() st.ahead, st.behind = int(a), -int(b) elif it[:2] in ("1 ", "2 "): f = it.split(" ", 9 if it[0] == "2" else 8) xy, path = f[1], f[-1] if it[0] == "2": # rename/copy: the old path follows path = "%s -> %s" % (items[i], path) i += 1 if xy[0] != ".": st.staged.append((xy[0], path)) if xy[1] != ".": st.unstaged.append((xy[1], path)) elif it.startswith("u "): st.conflicts.append(it.split(" ", 10)[-1]) elif it.startswith("? "): st.untracked.append(it[2:]) return st
def diff(self, *paths, staged=False, rev=None, stat=False): """The diff text: working tree vs index, staged=True index vs HEAD, rev= against a revision (rev="a..b" between two). stat=True: --stat.""" args = ["diff", "--no-color", "--no-ext-diff"] + (["--stat"] if stat else []) + ( ["--cached"] if staged else []) + ([rev] if rev else []) return Out(self.run(*args + ["--"] + list(paths)))
def changes(self, *paths, rev=None, staged=False): """[(status, path)] as git diff --name-status (renames detected).""" args = ["diff", "--name-status", "-z", "-M"] + (["--cached"] if staged else []) + ( [rev] if rev else []) + ["--"] + list(paths) f = [x for x in _git(self.cwd, args, text=False).decode("utf-8", "surrogateescape") .split("\0")] out, i = Lines(), 0 while i < len(f) and f[i]: s = f[i] if s[0] in "RC": out.append("%s\t%s -> %s" % (s, f[i + 1], f[i + 2])) i += 3 else: out.append("%s\t%s" % (s, f[i + 1])) i += 2 return out
def blame(self, path, start=None, end=None, rev=None): args = ["blame", "--porcelain"] + (["-L", "%s,%s" % (start, end or "")] if start else []) \ + ([rev] if rev else []) + ["--", path] lines = self.run(*args).split("\n") info, out, i = {}, Blame(), 0 while i < len(lines): head = lines[i].split(" ") if not HASH.match(head[0]): i += 1 continue sha, lineno = head[0], int(head[2]) meta = info.setdefault(sha, {}) i += 1 while i < len(lines) and not lines[i].startswith("\t"): k, _, v = lines[i].partition(" ") meta[k] = v i += 1 text = lines[i][1:] if i < len(lines) else "" i += 1 when = _when("%s %s" % (meta["author-time"], meta.get("author-tz", "+0000"))) \ if "author-time" in meta else None out.append(BlameLine((lineno, sha, meta.get("author", "?"), when, text))) return out
def config(self, key=None): """Effective config (system, global, repo): a dict of lists, or one key's last value (None if unset).""" if key: r = self.run("config", "--get", key, check=False) return r.strip() if r.ok else None out = {} for item in _git(self.cwd, ["config", "-z", "--list"], text=False).decode( "utf-8", "replace").split("\0"): if item: k, _, v = item.partition("\n") out.setdefault(k, []).append(v) return out
def add(self, *paths): return self.run("add", "--", *paths)
def commit(self, message, *paths, all=False, amend=False): """Commit the staged changes (all=True: every tracked change; paths: only those) with this message, passed on stdin so nothing needs quoting. Returns the new Commit, with its files.""" args = ["commit", "-F", "-"] + (["-a"] if all else []) + ( ["--amend"] if amend else []) + (["--"] + list(paths) if paths else []) self.run(*args, input=message) return self.show("HEAD")
def at(self, path): return repo(os.path.join(self.root or self.gitdir, os.path.expanduser(str(path))))
# -- module level: the repo of the session's cwd --------------------------------------_REPOS = {}
def repo(path="."): """The Repo containing path (cached, so pack indexes are read once).""" r = Repo(path) key = (r.gitdir, r.root) if key not in _REPOS: _REPOS[key] = r return _REPOS[key]
at = repo
def _here(name): def f(*a, **kw): return getattr(repo(), name)(*a, **kw) f.__name__ = name f.__doc__ = getattr(Repo, name).__doc__ return f
for _n in ("run", "status", "log", "show", "diff", "changes", "cat", "ls", "blame", "head", "branch", "branches", "tags", "resolve", "config", "add", "commit", "get", "refs", "obj"): globals()[_n] = _here(_n)
def root(): return repo().root
class _CallableModule(types.ModuleType): """git("log --oneline -3") runs git in the cwd's repo."""
def __call__(self, *args, **kw): if len(args) == 1 and isinstance(args[0], str): args = shlex.split(args[0]) try: r = repo() return r.run(*args, **kw) except GitError: # not in a repo: init, clone, --version... return _git(None, args, kw.get("input"), kw.get("check", True), kw.get("timeout", 120))
sys.modules[__name__].__class__ = _CallableModule