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
  14
  15 Path = str  # eg: "/home/user/.cache/git-cache/v1"
  16 Repo = str  # eg: "https://github.com/NixOS/nixpkgs.git"
  17 Ref = str   # eg: "master" or "v1.0.0"
  18 Rev = str   # eg: "53a27350551844e1ed1a9257690294767389ef0d"
  19
  20
  21 def git_cachedir(repo: Repo) -> Path:
  22     # Use xdg module when it's less painful to have as a dependency
  23     XDG_CACHE_HOME = Path(
  24         os.environ.get('XDG_CACHE_HOME', os.path.expanduser('~/.cache')))
  25
  26     return Path(os.path.join(
  27         XDG_CACHE_HOME,
  28         'git-cache/v1',
  29         hashlib.sha256(repo.encode()).hexdigest()))
  30
  31
  32 def verify_ancestry(repo: Repo, ref: Ref, rev: Rev) -> None:
  33     cachedir = git_cachedir(repo)
  34     logging.debug('Verifying rev %s is an ancestor of ref "%s"', rev, ref)
  35     subprocess.run(['git', '-C', cachedir, 'merge-base', '--is-ancestor',
  36                     rev, ref], check=True)
  37
  38
  39 def fetch(repo: Repo, ref: Ref) -> Tuple[Path, Rev]:
  40     cachedir = git_cachedir(repo)
  41     if not os.path.exists(cachedir):
  42         logging.debug("Initializing git repo")
  43         subprocess.run(['git', 'init', '--bare', cachedir], check=True)
  44
  45     logging.debug('Fetching ref "%s" from %s', ref, repo)
  46     # We don't use --force here because we want to abort and freak out if forced
  47     # updates are happening.
  48     subprocess.run(['git', '-C', cachedir, 'fetch', repo,
  49                     '%s:%s' % (ref, ref)], check=True)
  50
  51     with open(os.path.join(cachedir, 'refs', 'heads', ref)) as rev_file:
  52         rev = Rev(rev_file.read(999).strip())
  53     verify_ancestry(repo, ref, rev)
  54
  55     return cachedir, rev
  56
  57
  58 def ensure_rev_available(repo: Repo, ref: Ref, rev: Rev) -> Path:
  59     cachedir = git_cachedir(repo)
  60     if os.path.exists(cachedir):
  61         logging.debug('Checking if we already have rev %s', rev)
  62         process = subprocess.run(
  63             ['git', '-C', cachedir, 'cat-file', '-e', rev], check=False)
  64         if process.returncode == 0:
  65             logging.debug('We already have rev %s', rev)
  66             verify_ancestry(repo, ref, rev)
  67             return cachedir
  68         if process.returncode != 1:
  69             raise Exception(
  70                 'Could not test for presence of rev %s.  Is cache dir "%s" messed up?' %
  71                 (rev, cachedir))
  72
  73     logging.debug(
  74         'We do not have rev %s.  We will fetch ref "%s" and hope it appears.',
  75         rev, ref)
  76     fetch(repo, ref)
  77     logging.debug('Verifying that fetch retrieved rev %s', rev)
  78     subprocess.run(['git', '-C', cachedir, 'cat-file', '-e', rev], check=True)
  79
  80     return cachedir
  81
  82
  83 def _main() -> None:
  84     if len(sys.argv) == 3:
  85         print('{1} {0}'.format(*fetch(Repo(sys.argv[1]), Ref(sys.argv[2]))))
  86     elif len(sys.argv) == 4:
  87         print(ensure_rev_available(
  88             Repo(sys.argv[1]), Ref(sys.argv[2]), Rev(sys.argv[3])))
  89     else:
  90         usage = '''usage: git-cache repo ref [rev]
  91 example: git-cache https://github.com/NixOS/nixpkgs.git master'''
  92         print(usage, file=sys.stderr)
  93         sys.exit(1)