git.scottworley.com Git - git-cache/blob - git_cache.py

   1 # It would be nice if we could share the nix git cache, but as of the
   2 # time of writing it is transitioning from gitv2 (deprecated) to gitv3
   3 # (not ready yet), and trying to straddle them both is too far into nix
   4 # implementation details for my comfort.  So we re-implement here half of
   5 # nix's builtins.fetchGit.  :(
   6
   7 import hashlib
   8 import logging
   9 import os
  10 import subprocess
  11 import sys
  12
  13 from typing import Tuple, Union
  14
  15 import backoff
  16
  17 Path = str  # eg: "/home/user/.cache/git-cache/v1"
  18 Repo = str  # eg: "https://github.com/NixOS/nixpkgs.git"
  19 Ref = str   # eg: "master" or "v1.0.0"
  20 Rev = str   # eg: "53a27350551844e1ed1a9257690294767389ef0d"
  21 RefOrRev = Union[Ref, Rev]
  22
  23
  24 def git_cachedir(repo: Repo) -> Path:
  25     # Use xdg module when it's less painful to have as a dependency
  26     XDG_CACHE_HOME = Path(
  27         os.environ.get('XDG_CACHE_HOME', os.path.expanduser('~/.cache')))
  28
  29     return Path(os.path.join(
  30         XDG_CACHE_HOME,
  31         'git-cache/v1',
  32         hashlib.sha256(repo.encode()).hexdigest()))
  33
  34
  35 def is_ancestor(repo: Repo, descendant: RefOrRev, ancestor: RefOrRev) -> bool:
  36     cachedir = git_cachedir(repo)
  37     logging.debug('Checking if %s is an ancestor of %s', ancestor, descendant)
  38     process = subprocess.run(['git',
  39                               '-C',
  40                               cachedir,
  41                               'merge-base',
  42                               '--is-ancestor',
  43                               ancestor,
  44                               descendant],
  45                              check=False)
  46     return process.returncode == 0
  47
  48
  49 def verify_ancestry(
  50         repo: Repo,
  51         descendant: RefOrRev,
  52         ancestor: RefOrRev) -> None:
  53     if not is_ancestor(repo, descendant, ancestor):
  54         raise Exception('%s is not an ancestor of %s' % (ancestor, descendant))
  55
  56
  57 @backoff.on_exception(
  58     backoff.expo,
  59     subprocess.CalledProcessError,
  60     max_time=lambda: int(os.environ.get('BACKOFF_MAX_TIME', '30')))
  61 def _git_fetch(cachedir: Path, repo: Repo, ref: Ref) -> None:
  62     # We don't use --force here because we want to abort and freak out if forced
  63     # updates are happening.
  64     subprocess.run(['git', '-C', cachedir, 'fetch', repo,
  65                     '%s:%s' % (ref, ref)], check=True)
  66
  67
  68 def fetch(repo: Repo, ref: Ref) -> Tuple[Path, Rev]:
  69     cachedir = git_cachedir(repo)
  70     if not os.path.exists(cachedir):
  71         logging.debug("Initializing git repo")
  72         subprocess.run(['git', 'init', '--bare', cachedir],
  73                        check=True, stdout=sys.stderr)
  74
  75     logging.debug('Fetching ref "%s" from %s', ref, repo)
  76     _git_fetch(cachedir, repo, ref)
  77
  78     with open(os.path.join(cachedir, 'refs', 'heads', ref)) as rev_file:
  79         rev = Rev(rev_file.read(999).strip())
  80     verify_ancestry(repo, ref, rev)
  81
  82     return cachedir, rev
  83
  84
  85 def ensure_rev_available(repo: Repo, ref: Ref, rev: Rev) -> Path:
  86     cachedir = git_cachedir(repo)
  87     if os.path.exists(cachedir) and is_ancestor(repo, ref, rev):
  88         return cachedir
  89
  90     logging.debug(
  91         'We do not have rev %s.  We will fetch ref "%s" and hope it appears.',
  92         rev, ref)
  93     fetch(repo, ref)
  94     logging.debug('Verifying that fetch retrieved rev %s', rev)
  95     subprocess.run(['git', '-C', cachedir, 'cat-file', '-e', rev], check=True)
  96     verify_ancestry(repo, ref, rev)
  97
  98     return cachedir
  99
 100
 101 def _main() -> None:
 102     if len(sys.argv) == 3:
 103         print('{1} {0}'.format(*fetch(Repo(sys.argv[1]), Ref(sys.argv[2]))))
 104     elif len(sys.argv) == 4:
 105         print(ensure_rev_available(
 106             Repo(sys.argv[1]), Ref(sys.argv[2]), Rev(sys.argv[3])))
 107     else:
 108         usage = '''usage: git-cache repo ref [rev]
 109 example: git-cache https://github.com/NixOS/nixpkgs.git master'''
 110         print(usage, file=sys.stderr)
 111         sys.exit(1)