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