Something went wrong. Try again.
What happens if I push all of nixpkgs
Something went wrong. Try again.
16 kB · 468 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469#!/usr/bin/env nix-shell#!nix-shell update-octave-shell.nix -i python3
"""Update a Octave package expression by passing in the `.nix` file, or the directory containing it.You can pass in multiple files or paths.
You'll likely want to use`` $ ./update-octave-libraries ../../pkgs/development/octave-modules/**/default.nix``to update all non-pinned libraries in that folder."""
import argparseimport osimport pathlibimport reimport requestsimport yamlfrom concurrent.futures import ThreadPoolExecutor as Poolfrom packaging.version import Version as _Versionfrom packaging.version import InvalidVersionfrom packaging.specifiers import SpecifierSetimport collectionsimport subprocessimport tempfile
INDEX = "https://raw.githubusercontent.com/gnu-octave/packages/main/packages""""url of Octave packages' source on GitHub"""
EXTENSIONS = ['tar.gz', 'tar.bz2', 'tar', 'zip']"""Permitted file extensions. These are evaluated from left to right and the first occurance is returned."""
PRERELEASES = False
GIT = "git"
NIXPGKS_ROOT = subprocess.check_output(["git", "rev-parse", "--show-toplevel"]).decode('utf-8').strip()
import logginglogging.basicConfig(level=logging.INFO)
class Version(_Version, collections.abc.Sequence):
def __init__(self, version): super().__init__(version) # We cannot use `str(Version(0.04.21))` because that becomes `0.4.21` # https://github.com/avian2/unidecode/issues/13#issuecomment-354538882 self.raw_version = version
def __getitem__(self, i): return self._version.release[i]
def __len__(self): return len(self._version.release)
def __iter__(self): yield from self._version.release
def _get_values(attribute, text): """Match attribute in text and return all matches.
:returns: List of matches. """ regex = '{}\s+=\s+"(.*)";'.format(attribute) regex = re.compile(regex) values = regex.findall(text) return values
def _get_unique_value(attribute, text): """Match attribute in text and return unique match.
:returns: Single match. """ values = _get_values(attribute, text) n = len(values) if n > 1: raise ValueError("found too many values for {}".format(attribute)) elif n == 1: return values[0] else: raise ValueError("no value found for {}".format(attribute))
def _get_line_and_value(attribute, text): """Match attribute in text. Return the line and the value of the attribute.""" regex = '({}\s+=\s+"(.*)";)'.format(attribute) regex = re.compile(regex) value = regex.findall(text) n = len(value) if n > 1: raise ValueError("found too many values for {}".format(attribute)) elif n == 1: return value[0] else: raise ValueError("no value found for {}".format(attribute))
def _replace_value(attribute, value, text): """Search and replace value of attribute in text.""" old_line, old_value = _get_line_and_value(attribute, text) new_line = old_line.replace(old_value, value) new_text = text.replace(old_line, new_line) return new_text
def _fetch_page(url): r = requests.get(url) if r.status_code == requests.codes.ok: return list(yaml.safe_load_all(r.content))[0] else: raise ValueError("request for {} failed".format(url))
def _fetch_github(url): headers = {} token = os.environ.get('GITHUB_API_TOKEN') if token: headers["Authorization"] = f"token {token}" r = requests.get(url, headers=headers)
if r.status_code == requests.codes.ok: return r.json() else: raise ValueError("request for {} failed".format(url))
SEMVER = { 'major' : 0, 'minor' : 1, 'patch' : 2,}
def _determine_latest_version(current_version, target, versions): """Determine latest version, given `target`, returning the more recent version. """ current_version = Version(current_version)
def _parse_versions(versions): for v in versions: try: yield Version(v) except InvalidVersion: pass
versions = _parse_versions(versions)
index = SEMVER[target]
ceiling = list(current_version[0:index]) if len(ceiling) == 0: ceiling = None else: ceiling[-1]+=1 ceiling = Version(".".join(map(str, ceiling)))
# We do not want prereleases versions = SpecifierSet(prereleases=PRERELEASES).filter(versions)
if ceiling is not None: versions = SpecifierSet(f"<{ceiling}").filter(versions)
return (max(sorted(versions))).raw_version
def _get_latest_version_octave_packages(package, extension, current_version, target): """Get latest version and hash from Octave Packages.""" url = "{}/{}.yaml".format(INDEX, package) yaml = _fetch_page(url)
versions = list(map(lambda pv: pv['id'], yaml['versions'])) version = _determine_latest_version(current_version, target, versions)
try: releases = [v if v['id'] == version else None for v in yaml['versions']] except KeyError as e: raise KeyError('Could not find version {} for {}'.format(version, package)) from e for release in releases: if release['url'].endswith(extension): sha256 = release['sha256'] break else: sha256 = None return version, sha256, None
def _get_latest_version_github(package, extension, current_version, target): def strip_prefix(tag): return re.sub("^[^0-9]*", "", tag)
def get_prefix(string): matches = re.findall(r"^([^0-9]*)", string) return next(iter(matches), "")
# when invoked as an updateScript, UPDATE_NIX_ATTR_PATH will be set # this allows us to work with packages which live outside of octave-modules attr_path = os.environ.get("UPDATE_NIX_ATTR_PATH", f"octavePackages.{package}") try: homepage = subprocess.check_output( ["nix", "eval", "-f", f"{NIXPGKS_ROOT}/default.nix", "--raw", f"{attr_path}.src.meta.homepage"])\ .decode('utf-8') except Exception as e: raise ValueError(f"Unable to determine homepage: {e}") owner_repo = homepage[len("https://github.com/"):] # remove prefix owner, repo = owner_repo.split("/")
url = f"https://api.github.com/repos/{owner}/{repo}/releases" all_releases = _fetch_github(url) releases = list(filter(lambda x: not x['prerelease'], all_releases))
if len(releases) == 0: raise ValueError(f"{homepage} does not contain any stable releases")
versions = map(lambda x: strip_prefix(x['tag_name']), releases) version = _determine_latest_version(current_version, target, versions)
release = next(filter(lambda x: strip_prefix(x['tag_name']) == version, releases)) prefix = get_prefix(release['tag_name']) try: sha256 = subprocess.check_output(["nix-prefetch-url", "--type", "sha256", "--unpack", f"{release['tarball_url']}"], stderr=subprocess.DEVNULL)\ .decode('utf-8').strip() except: # this may fail if they have both a branch and a tag of the same name, attempt tag name tag_url = str(release['tarball_url']).replace("tarball","tarball/refs/tags") sha256 = subprocess.check_output(["nix-prefetch-url", "--type", "sha256", "--unpack", tag_url], stderr=subprocess.DEVNULL)\ .decode('utf-8').strip()
return version, sha256, prefix
def _get_latest_version_git(package, extension, current_version, target): """NOTE: Unimplemented!""" # attr_path = os.environ.get("UPDATE_NIX_ATTR_PATH", f"octavePackages.{package}") # try: # download_url = subprocess.check_output( # ["nix", "--extra-experimental-features", "nix-command", "eval", "-f", f"{NIXPGKS_ROOT}/default.nix", "--raw", f"{attr_path}.src.url"])\ # .decode('utf-8') # except Exception as e: # raise ValueError(f"Unable to determine download link: {e}")
# with tempfile.TemporaryDirectory(prefix=attr_path) as new_clone_location: # subprocess.run(["git", "clone", download_url, new_clone_location]) # newest_commit = subprocess.check_output( # ["git" "rev-parse" "$(git branch -r)" "|" "tail" "-n" "1"]).decode('utf-8') pass
FETCHERS = { 'fetchFromGitHub' : _get_latest_version_github, 'fetchurl' : _get_latest_version_octave_packages, 'fetchgit' : _get_latest_version_git,}
DEFAULT_SETUPTOOLS_EXTENSION = 'tar.gz'
FORMATS = { 'setuptools' : DEFAULT_SETUPTOOLS_EXTENSION,}
def _determine_fetcher(text): # Count occurrences of fetchers. nfetchers = sum(text.count('src = {}'.format(fetcher)) for fetcher in FETCHERS.keys()) if nfetchers == 0: raise ValueError("no fetcher.") elif nfetchers > 1: raise ValueError("multiple fetchers.") else: # Then we check which fetcher to use. for fetcher in FETCHERS.keys(): if 'src = {}'.format(fetcher) in text: return fetcher
def _determine_extension(text, fetcher): """Determine what extension is used in the expression.
If we use: - fetchPypi, we check if format is specified. - fetchurl, we determine the extension from the url. - fetchFromGitHub we simply use `.tar.gz`. """ if fetcher == 'fetchurl': url = _get_unique_value('url', text) extension = os.path.splitext(url)[1]
elif fetcher == 'fetchFromGitHub' or fetcher == 'fetchgit': if "fetchSubmodules" in text: raise ValueError("fetchFromGitHub fetcher doesn't support submodules") extension = "tar.gz"
return extension
def _update_package(path, target):
# Read the expression with open(path, 'r') as f: text = f.read()
# Determine pname. Many files have more than one pname pnames = _get_values('pname', text)
# Determine version. version = _get_unique_value('version', text)
# First we check how many fetchers are mentioned. fetcher = _determine_fetcher(text)
extension = _determine_extension(text, fetcher)
# Attempt a fetch using each pname, e.g. backports-zoneinfo vs backports.zoneinfo successful_fetch = False for pname in pnames: if fetcher == "fetchgit": logging.warning(f"You must update {pname} MANUALLY!") return { 'path': path, 'target': target, 'pname': pname, 'old_version': version, 'new_version': version } try: new_version, new_sha256, prefix = FETCHERS[fetcher](pname, extension, version, target) successful_fetch = True break except ValueError: continue
if not successful_fetch: raise ValueError(f"Unable to find correct package using these pnames: {pnames}")
if new_version == version: logging.info("Path {}: no update available for {}.".format(path, pname)) return False elif Version(new_version) <= Version(version): raise ValueError("downgrade for {}.".format(pname)) if not new_sha256: raise ValueError("no file available for {}.".format(pname))
text = _replace_value('version', new_version, text) # hashes from pypi are 16-bit encoded sha256's, normalize it to sri to avoid merge conflicts # sri hashes have been the default format since nix 2.4+ sri_hash = subprocess.check_output(["nix", "--extra-experimental-features", "nix-command", "hash", "to-sri", "--type", "sha256", new_sha256]).decode('utf-8').strip()
# fetchers can specify a sha256, or a sri hash try: text = _replace_value('sha256', sri_hash, text) except ValueError: text = _replace_value('hash', sri_hash, text)
if fetcher == 'fetchFromGitHub': # in the case of fetchFromGitHub, it's common to see `rev = version;` or `rev = "v${version}";` # in which no string value is meant to be substituted. However, we can just overwrite the previous value. regex = '(rev\s+=\s+[^;]*;)' regex = re.compile(regex) matches = regex.findall(text) n = len(matches)
if n == 0: raise ValueError("Unable to find rev value for {}.".format(pname)) else: # forcefully rewrite rev, incase tagging conventions changed for a release match = matches[0] text = text.replace(match, f'rev = "refs/tags/{prefix}${{version}}";') # incase there's no prefix, just rewrite without interpolation text = text.replace('"${version}";', 'version;')
with open(path, 'w') as f: f.write(text)
logging.info("Path {}: updated {} from {} to {}".format(path, pname, version, new_version))
result = { 'path' : path, 'target': target, 'pname': pname, 'old_version' : version, 'new_version' : new_version, #'fetcher' : fetcher, }
return result
def _update(path, target):
# We need to read and modify a Nix expression. if os.path.isdir(path): path = os.path.join(path, 'default.nix')
# If a default.nix does not exist, we quit. if not os.path.isfile(path): logging.info("Path {}: does not exist.".format(path)) return False
# If file is not a Nix expression, we quit. if not path.endswith(".nix"): logging.info("Path {}: does not end with `.nix`.".format(path)) return False
try: return _update_package(path, target) except ValueError as e: logging.warning("Path {}: {}".format(path, e)) return False
def _commit(path, pname, old_version, new_version, pkgs_prefix="octave: ", **kwargs): """Commit result. """
msg = f'{pkgs_prefix}{pname}: {old_version} -> {new_version}'
try: subprocess.check_call([GIT, 'add', path]) subprocess.check_call([GIT, 'commit', '-m', msg]) except subprocess.CalledProcessError as e: subprocess.check_call([GIT, 'checkout', path]) raise subprocess.CalledProcessError(f'Could not commit {path}') from e
return True
def main():
epilog = """environment variables: GITHUB_API_TOKEN\tGitHub API token used when updating github packages """ parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter, epilog=epilog) parser.add_argument('package', type=str, nargs='+') parser.add_argument('--target', type=str, choices=SEMVER.keys(), default='major') parser.add_argument('--commit', action='store_true', help='Create a commit for each package update') parser.add_argument('--use-pkgs-prefix', action='store_true', help='Use octavePackages.${pname}: instead of octave: ${pname}: when making commits')
args = parser.parse_args() target = args.target
packages = list(map(os.path.abspath, args.package))
logging.info("Updating packages...")
# Use threads to update packages concurrently with Pool() as p: results = list(filter(bool, p.map(lambda pkg: _update(pkg, target), packages)))
logging.info("Finished updating packages.")
commit_options = {} if args.use_pkgs_prefix: logging.info("Using octavePackages. prefix for commits") commit_options["pkgs_prefix"] = "octavePackages."
# Commits are created sequentially. if args.commit: logging.info("Committing updates...") # list forces evaluation list(map(lambda x: _commit(**x, **commit_options), results)) logging.info("Finished committing updates")
count = len(results) logging.info("{} package(s) updated".format(count))
if __name__ == '__main__': main()