_setenv()

[git-cache] / git_cache.py

# It would be nice if we could share the nix git cache, but as of the
# time of writing it is transitioning from gitv2 (deprecated) to gitv3
# (not ready yet), and trying to straddle them both is too far into nix
# implementation details for my comfort.  So we re-implement here half of
# nix's builtins.fetchGit.  :(

import hashlib
import logging
import os
import subprocess
import sys

from typing import Tuple, Union

import backoff

Path = str  # eg: "/home/user/.cache/git-cache/v1"
Repo = str  # eg: "https://github.com/NixOS/nixpkgs.git"
Ref = str   # eg: "master" or "v1.0.0"
Rev = str   # eg: "53a27350551844e1ed1a9257690294767389ef0d"
RefOrRev = Union[Ref, Rev]


def git_cachedir(repo: Repo) -> Path:
    # Use xdg module when it's less painful to have as a dependency
    XDG_CACHE_HOME = Path(
        os.environ.get('XDG_CACHE_HOME', os.path.expanduser('~/.cache')))

    return Path(os.path.join(
        XDG_CACHE_HOME,
        'git-cache/v1',
        hashlib.sha256(repo.encode()).hexdigest()))


def is_ancestor(repo: Repo, descendant: RefOrRev, ancestor: RefOrRev) -> bool:
    cachedir = git_cachedir(repo)
    logging.debug('Checking if %s is an ancestor of %s', ancestor, descendant)
    process = subprocess.run(['git',
                              '-C',
                              cachedir,
                              'merge-base',
                              '--is-ancestor',
                              ancestor,
                              descendant],
                             check=False)
    return process.returncode == 0


def verify_ancestry(
        repo: Repo,
        descendant: RefOrRev,
        ancestor: RefOrRev) -> None:
    if not is_ancestor(repo, descendant, ancestor):
        raise Exception('%s is not an ancestor of %s' % (ancestor, descendant))


@backoff.on_exception(
    backoff.expo,
    subprocess.CalledProcessError,
    max_time=lambda: int(os.environ.get('BACKOFF_MAX_TIME', '30')))
def _git_fetch(cachedir: Path, repo: Repo, ref: Ref) -> None:
    # We don't use --force here because we want to abort and freak out if forced
    # updates are happening.
    subprocess.run(['git', '-C', cachedir, 'fetch', repo,
                    '%s:%s' % (ref, ref)], check=True)


def fetch(repo: Repo, ref: Ref) -> Tuple[Path, Rev]:
    cachedir = git_cachedir(repo)
    if not os.path.exists(cachedir):
        logging.debug("Initializing git repo")
        subprocess.run(['git', 'init', '--bare', cachedir],
                       check=True, stdout=sys.stderr)

    logging.debug('Fetching ref "%s" from %s', ref, repo)
    _git_fetch(cachedir, repo, ref)

    with open(os.path.join(cachedir, 'refs', 'heads', ref)) as rev_file:
        rev = Rev(rev_file.read(999).strip())
    verify_ancestry(repo, ref, rev)

    return cachedir, rev


def ensure_rev_available(repo: Repo, ref: Ref, rev: Rev) -> Path:
    cachedir = git_cachedir(repo)
    if os.path.exists(cachedir) and is_ancestor(repo, ref, rev):
        return cachedir

    logging.debug(
        'We do not have rev %s.  We will fetch ref "%s" and hope it appears.',
        rev, ref)
    fetch(repo, ref)
    logging.debug('Verifying that fetch retrieved rev %s', rev)
    subprocess.run(['git', '-C', cachedir, 'cat-file', '-e', rev], check=True)
    verify_ancestry(repo, ref, rev)

    return cachedir


def _main() -> None:
    if len(sys.argv) == 3:
        print('{1} {0}'.format(*fetch(Repo(sys.argv[1]), Ref(sys.argv[2]))))
    elif len(sys.argv) == 4:
        print(ensure_rev_available(
            Repo(sys.argv[1]), Ref(sys.argv[2]), Rev(sys.argv[3])))
    else:
        usage = '''usage: git-cache repo ref [rev]
example: git-cache https://github.com/NixOS/nixpkgs.git master'''
        print(usage, file=sys.stderr)
        sys.exit(1)