BoomMicrophone/nixpkgs-suyu
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

pluginupdate.py: format with black, fix comments

Browse source
This commit is contained in:
figsoda 2023-08-12 11:41:04 -04:00
parent 32f6cfaae5
commit 90c4482d07
1 changed files with 103 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

178
maintainers/scripts/pluginupdate.py
View file

@ -4,20 +4,20 @@
# - maintainers/scripts/update-luarocks-packages # - maintainers/scripts/update-luarocks-packages
# format: # format:
# $ nix run nixpkgs.python3Packages.black -c black update.py # $ nix run nixpkgs#black maintainers/scripts/pluginupdate.py
# type-check: # type-check:
# $ nix run nixpkgs.python3Packages.mypy -c mypy update.py # $ nix run nixpkgs#python3.pkgs.mypy maintainers/scripts/pluginupdate.py
# linted: # linted:
# $ nix run nixpkgs.python3Packages.flake8 -c flake8 --ignore E501,E265 update.py # $ nix run nixpkgs#python3.pkgs.flake8 -- --ignore E501,E265 maintainers/scripts/pluginupdate.py
import argparse import argparse
import csv import csv
import functools import functools
import http import http
import json import json
import logging
import os import os
import subprocess import subprocess
import logging
import sys import sys
import time import time
import traceback import traceback
@ -25,14 +25,14 @@ import urllib.error
import urllib.parse import urllib.parse
import urllib.request import urllib.request
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from dataclasses import asdict, dataclass
from datetime import datetime from datetime import datetime
from functools import wraps from functools import wraps
from multiprocessing.dummy import Pool from multiprocessing.dummy import Pool
from pathlib import Path from pathlib import Path
from typing import Dict, List, Optional, Tuple, Union, Any, Callable
from urllib.parse import urljoin, urlparse
from tempfile import NamedTemporaryFile from tempfile import NamedTemporaryFile
from dataclasses import dataclass, asdict from typing import Any, Callable, Dict, List, Optional, Tuple, Union
from urllib.parse import urljoin, urlparse
import git import git
@ -41,12 +41,13 @@ ATOM_LINK = "{http://www.w3.org/2005/Atom}link" # "
ATOM_UPDATED = "{http://www.w3.org/2005/Atom}updated" # " ATOM_UPDATED = "{http://www.w3.org/2005/Atom}updated" # "
LOG_LEVELS = { LOG_LEVELS = {
logging.getLevelName(level): level for level in [ logging.getLevelName(level): level
logging.DEBUG, logging.INFO, logging.WARN, logging.ERROR ] for level in [logging.DEBUG, logging.INFO, logging.WARN, logging.ERROR]
} }
log = logging.getLogger() log = logging.getLogger()
def retry(ExceptionToCheck: Any, tries: int = 4, delay: float = 3, backoff: float = 2): def retry(ExceptionToCheck: Any, tries: int = 4, delay: float = 3, backoff: float = 2):
"""Retry calling the decorated function using an exponential backoff. """Retry calling the decorated function using an exponential backoff.
http://www.saltycrane.com/blog/2009/11/trying-out-retry-decorator-python/ http://www.saltycrane.com/blog/2009/11/trying-out-retry-decorator-python/
@ -77,6 +78,7 @@ def retry(ExceptionToCheck: Any, tries: int = 4, delay: float = 3, backoff: floa
return deco_retry return deco_retry
@dataclass @dataclass
class FetchConfig: class FetchConfig:
proc: int proc: int
@ -91,22 +93,21 @@ def make_request(url: str, token=None) -> urllib.request.Request:
# a dictionary of plugins and their new repositories # a dictionary of plugins and their new repositories
Redirects = Dict['PluginDesc', 'Repo'] Redirects = Dict["PluginDesc", "Repo"]
class Repo: class Repo:
def __init__( def __init__(self, uri: str, branch: str) -> None:
self, uri: str, branch: str
) -> None:
self.uri = uri self.uri = uri
'''Url to the repo''' """Url to the repo"""
self._branch = branch self._branch = branch
# Redirect is the new Repo to use # Redirect is the new Repo to use
self.redirect: Optional['Repo'] = None self.redirect: Optional["Repo"] = None
self.token = "dummy_token" self.token = "dummy_token"
@property @property
def name(self): def name(self):
return self.uri.split('/')[-1] return self.uri.split("/")[-1]
@property @property
def branch(self): def branch(self):
@ -114,6 +115,7 @@ class Repo:
def __str__(self) -> str: def __str__(self) -> str:
return f"{self.uri}" return f"{self.uri}"
def __repr__(self) -> str: def __repr__(self) -> str:
return f"Repo({self.name}, {self.uri})" return f"Repo({self.name}, {self.uri})"
@ -125,9 +127,9 @@ class Repo:
def latest_commit(self) -> Tuple[str, datetime]: def latest_commit(self) -> Tuple[str, datetime]:
log.debug("Latest commit") log.debug("Latest commit")
loaded = self._prefetch(None) loaded = self._prefetch(None)
updated = datetime.strptime(loaded['date'], "%Y-%m-%dT%H:%M:%S%z") updated = datetime.strptime(loaded["date"], "%Y-%m-%dT%H:%M:%S%z")
return loaded['rev'], updated return loaded["rev"], updated
def _prefetch(self, ref: Optional[str]): def _prefetch(self, ref: Optional[str]):
cmd = ["nix-prefetch-git", "--quiet", "--fetch-submodules", self.uri] cmd = ["nix-prefetch-git", "--quiet", "--fetch-submodules", self.uri]
@ -144,23 +146,23 @@ class Repo:
return loaded["sha256"] return loaded["sha256"]
def as_nix(self, plugin: "Plugin") -> str: def as_nix(self, plugin: "Plugin") -> str:
return f'''fetchgit {{ return f"""fetchgit {{
url = "{self.uri}"; url = "{self.uri}";
rev = "{plugin.commit}"; rev = "{plugin.commit}";
sha256 = "{plugin.sha256}"; sha256 = "{plugin.sha256}";
}}''' }}"""
class RepoGitHub(Repo): class RepoGitHub(Repo):
def __init__( def __init__(self, owner: str, repo: str, branch: str) -> None:
self, owner: str, repo: str, branch: str
) -> None:
self.owner = owner self.owner = owner
self.repo = repo self.repo = repo
self.token = None self.token = None
'''Url to the repo''' """Url to the repo"""
super().__init__(self.url(""), branch) super().__init__(self.url(""), branch)
log.debug("Instantiating github repo owner=%s and repo=%s", self.owner, self.repo) log.debug(
"Instantiating github repo owner=%s and repo=%s", self.owner, self.repo
)
@property @property
def name(self): def name(self):
@ -213,7 +215,6 @@ class RepoGitHub(Repo):
new_repo = RepoGitHub(owner=new_owner, repo=new_name, branch=self.branch) new_repo = RepoGitHub(owner=new_owner, repo=new_name, branch=self.branch)
self.redirect = new_repo self.redirect = new_repo
def prefetch(self, commit: str) -> str: def prefetch(self, commit: str) -> str:
if self.has_submodules(): if self.has_submodules():
sha256 = super().prefetch(commit) sha256 = super().prefetch(commit)
@ -233,12 +234,12 @@ class RepoGitHub(Repo):
else: else:
submodule_attr = "" submodule_attr = ""
return f'''fetchFromGitHub {{ return f"""fetchFromGitHub {{
owner = "{self.owner}"; owner = "{self.owner}";
repo = "{self.repo}"; repo = "{self.repo}";
rev = "{plugin.commit}"; rev = "{plugin.commit}";
sha256 = "{plugin.sha256}";{submodule_attr} sha256 = "{plugin.sha256}";{submodule_attr}
}}''' }}"""
@dataclass(frozen=True) @dataclass(frozen=True)
@ -258,15 +259,14 @@ class PluginDesc:
return self.repo.name < other.repo.name return self.repo.name < other.repo.name
@staticmethod @staticmethod
def load_from_csv(config: FetchConfig, row: Dict[str, str]) -> 'PluginDesc': def load_from_csv(config: FetchConfig, row: Dict[str, str]) -> "PluginDesc":
branch = row["branch"] branch = row["branch"]
repo = make_repo(row['repo'], branch.strip()) repo = make_repo(row["repo"], branch.strip())
repo.token = config.github_token repo.token = config.github_token
return PluginDesc(repo, branch.strip(), row["alias"]) return PluginDesc(repo, branch.strip(), row["alias"])
@staticmethod @staticmethod
def load_from_string(config: FetchConfig, line: str) -> 'PluginDesc': def load_from_string(config: FetchConfig, line: str) -> "PluginDesc":
branch = "HEAD" branch = "HEAD"
alias = None alias = None
uri = line uri = line
@ -279,6 +279,7 @@ class PluginDesc:
repo.token = config.github_token repo.token = config.github_token
return PluginDesc(repo, branch.strip(), alias) return PluginDesc(repo, branch.strip(), alias)
@dataclass @dataclass
class Plugin: class Plugin:
name: str name: str
@ -302,23 +303,38 @@ class Plugin:
return copy return copy
def load_plugins_from_csv(config: FetchConfig, input_file: Path,) -> List[PluginDesc]: def load_plugins_from_csv(
config: FetchConfig,
input_file: Path,
) -> List[PluginDesc]:
log.debug("Load plugins from csv %s", input_file) log.debug("Load plugins from csv %s", input_file)
plugins = [] plugins = []
with open(input_file, newline='') as csvfile: with open(input_file, newline="") as csvfile:
log.debug("Writing into %s", input_file) log.debug("Writing into %s", input_file)
reader = csv.DictReader(csvfile,) reader = csv.DictReader(
csvfile,
)
for line in reader: for line in reader:
plugin = PluginDesc.load_from_csv(config, line) plugin = PluginDesc.load_from_csv(config, line)
plugins.append(plugin) plugins.append(plugin)
return plugins return plugins
def run_nix_expr(expr): def run_nix_expr(expr):
with CleanEnvironment() as nix_path: with CleanEnvironment() as nix_path:
cmd = ["nix", "eval", "--extra-experimental-features", cmd = [
"nix-command", "--impure", "--json", "--expr", expr, "nix",
"--nix-path", nix_path] "eval",
"--extra-experimental-features",
"nix-command",
"--impure",
"--json",
"--expr",
expr,
"--nix-path",
nix_path,
]
log.debug("Running command %s", " ".join(cmd)) log.debug("Running command %s", " ".join(cmd))
out = subprocess.check_output(cmd) out = subprocess.check_output(cmd)
data = json.loads(out) data = json.loads(out)
@ -349,7 +365,7 @@ class Editor:
self.nixpkgs_repo = None self.nixpkgs_repo = None
def add(self, args): def add(self, args):
'''CSV spec''' """CSV spec"""
log.debug("called the 'add' command") log.debug("called the 'add' command")
fetch_config = FetchConfig(args.proc, args.github_token) fetch_config = FetchConfig(args.proc, args.github_token)
editor = self editor = self
@ -357,23 +373,27 @@ class Editor:
log.debug("using plugin_line", plugin_line) log.debug("using plugin_line", plugin_line)
pdesc = PluginDesc.load_from_string(fetch_config, plugin_line) pdesc = PluginDesc.load_from_string(fetch_config, plugin_line)
log.debug("loaded as pdesc", pdesc) log.debug("loaded as pdesc", pdesc)
append = [ pdesc ] append = [pdesc]
editor.rewrite_input(fetch_config, args.input_file, editor.deprecated, append=append) editor.rewrite_input(
plugin, _ = prefetch_plugin(pdesc, ) fetch_config, args.input_file, editor.deprecated, append=append
)
plugin, _ = prefetch_plugin(
pdesc,
)
autocommit = not args.no_commit autocommit = not args.no_commit
if autocommit: if autocommit:
commit( commit(
editor.nixpkgs_repo, editor.nixpkgs_repo,
"{drv_name}: init at {version}".format( "{drv_name}: init at {version}".format(
drv_name=editor.get_drv_name(plugin.normalized_name), drv_name=editor.get_drv_name(plugin.normalized_name),
version=plugin.version version=plugin.version,
), ),
[args.outfile, args.input_file], [args.outfile, args.input_file],
) )
# Expects arguments generated by 'update' subparser # Expects arguments generated by 'update' subparser
def update(self, args ): def update(self, args):
'''CSV spec''' """CSV spec"""
print("the update member function should be overriden in subclasses") print("the update member function should be overriden in subclasses")
def get_current_plugins(self) -> List[Plugin]: def get_current_plugins(self) -> List[Plugin]:
@ -386,11 +406,11 @@ class Editor:
return plugins return plugins
def load_plugin_spec(self, config: FetchConfig, plugin_file) -> List[PluginDesc]: def load_plugin_spec(self, config: FetchConfig, plugin_file) -> List[PluginDesc]:
'''CSV spec''' """CSV spec"""
return load_plugins_from_csv(config, plugin_file) return load_plugins_from_csv(config, plugin_file)
def generate_nix(self, _plugins, _outfile: str): def generate_nix(self, _plugins, _outfile: str):
'''Returns nothing for now, writes directly to outfile''' """Returns nothing for now, writes directly to outfile"""
raise NotImplementedError() raise NotImplementedError()
def get_update(self, input_file: str, outfile: str, config: FetchConfig): def get_update(self, input_file: str, outfile: str, config: FetchConfig):
@ -414,7 +434,6 @@ class Editor:
return update return update
@property @property
def attr_path(self): def attr_path(self):
return self.name + "Plugins" return self.name + "Plugins"
@ -428,10 +447,11 @@ class Editor:
def create_parser(self): def create_parser(self):
common = argparse.ArgumentParser( common = argparse.ArgumentParser(
add_help=False, add_help=False,
description=(f""" description=(
f"""
Updates nix derivations for {self.name} plugins.\n Updates nix derivations for {self.name} plugins.\n
By default from {self.default_in} to {self.default_out}""" By default from {self.default_in} to {self.default_out}"""
) ),
) )
common.add_argument( common.add_argument(
"--input-names", "--input-names",
@ -464,26 +484,33 @@ class Editor:
Uses GITHUB_API_TOKEN environment variables as the default value.""", Uses GITHUB_API_TOKEN environment variables as the default value.""",
) )
common.add_argument( common.add_argument(
"--no-commit", "-n", action="store_true", default=False, "--no-commit",
help="Whether to autocommit changes" "-n",
action="store_true",
default=False,
help="Whether to autocommit changes",
) )
common.add_argument( common.add_argument(
"--debug", "-d", choices=LOG_LEVELS.keys(), "--debug",
"-d",
choices=LOG_LEVELS.keys(),
default=logging.getLevelName(logging.WARN), default=logging.getLevelName(logging.WARN),
help="Adjust log level" help="Adjust log level",
) )
main = argparse.ArgumentParser( main = argparse.ArgumentParser(
parents=[common], parents=[common],
description=(f""" description=(
f"""
Updates nix derivations for {self.name} plugins.\n Updates nix derivations for {self.name} plugins.\n
By default from {self.default_in} to {self.default_out}""" By default from {self.default_in} to {self.default_out}"""
) ),
) )
subparsers = main.add_subparsers(dest="command", required=False) subparsers = main.add_subparsers(dest="command", required=False)
padd = subparsers.add_parser( padd = subparsers.add_parser(
"add", parents=[], "add",
parents=[],
description="Add new plugin", description="Add new plugin",
add_help=False, add_help=False,
) )
@ -503,10 +530,12 @@ class Editor:
pupdate.set_defaults(func=self.update) pupdate.set_defaults(func=self.update)
return main return main
def run(self,): def run(
''' self,
):
"""
Convenience function Convenience function
''' """
parser = self.create_parser() parser = self.create_parser()
args = parser.parse_args() args = parser.parse_args()
command = args.command or "update" command = args.command or "update"
@ -519,8 +548,6 @@ class Editor:
getattr(self, command)(args) getattr(self, command)(args)
class CleanEnvironment(object): class CleanEnvironment(object):
def __enter__(self) -> str: def __enter__(self) -> str:
self.old_environ = os.environ.copy() self.old_environ = os.environ.copy()
@ -571,14 +598,15 @@ def print_download_error(plugin: PluginDesc, ex: Exception):
] ]
print("\n".join(tb_lines)) print("\n".join(tb_lines))
def check_results( def check_results(
results: List[Tuple[PluginDesc, Union[Exception, Plugin], Optional[Repo]]] results: List[Tuple[PluginDesc, Union[Exception, Plugin], Optional[Repo]]]
) -> Tuple[List[Tuple[PluginDesc, Plugin]], Redirects]: ) -> Tuple[List[Tuple[PluginDesc, Plugin]], Redirects]:
''' ''' """ """
failures: List[Tuple[PluginDesc, Exception]] = [] failures: List[Tuple[PluginDesc, Exception]] = []
plugins = [] plugins = []
redirects: Redirects = {} redirects: Redirects = {}
for (pdesc, result, redirect) in results: for pdesc, result, redirect in results:
if isinstance(result, Exception): if isinstance(result, Exception):
failures.append((pdesc, result)) failures.append((pdesc, result))
else: else:
@ -595,17 +623,18 @@ def check_results(
else: else:
print(f", {len(failures)} plugin(s) could not be downloaded:\n") print(f", {len(failures)} plugin(s) could not be downloaded:\n")
for (plugin, exception) in failures: for plugin, exception in failures:
print_download_error(plugin, exception) print_download_error(plugin, exception)
sys.exit(1) sys.exit(1)
def make_repo(uri: str, branch) -> Repo: def make_repo(uri: str, branch) -> Repo:
'''Instantiate a Repo with the correct specialization depending on server (gitub spec)''' """Instantiate a Repo with the correct specialization depending on server (gitub spec)"""
# dumb check to see if it's of the form owner/repo (=> github) or https://... # dumb check to see if it's of the form owner/repo (=> github) or https://...
res = urlparse(uri) res = urlparse(uri)
if res.netloc in [ "github.com", ""]: if res.netloc in ["github.com", ""]:
res = res.path.strip('/').split('/') res = res.path.strip("/").split("/")
repo = RepoGitHub(res[0], res[1], branch) repo = RepoGitHub(res[0], res[1], branch)
else: else:
repo = Repo(uri.strip(), branch) repo = Repo(uri.strip(), branch)
@ -676,7 +705,6 @@ def prefetch(
return (pluginDesc, e, None) return (pluginDesc, e, None)
def rewrite_input( def rewrite_input(
config: FetchConfig, config: FetchConfig,
input_file: Path, input_file: Path,
@ -685,12 +713,14 @@ def rewrite_input(
redirects: Redirects = {}, redirects: Redirects = {},
append: List[PluginDesc] = [], append: List[PluginDesc] = [],
): ):
plugins = load_plugins_from_csv(config, input_file,) plugins = load_plugins_from_csv(
config,
input_file,
)
plugins.extend(append) plugins.extend(append)
if redirects: if redirects:
cur_date_iso = datetime.now().strftime("%Y-%m-%d") cur_date_iso = datetime.now().strftime("%Y-%m-%d")
with open(deprecated, "r") as f: with open(deprecated, "r") as f:
deprecations = json.load(f) deprecations = json.load(f)
@ -710,8 +740,8 @@ def rewrite_input(
with open(input_file, "w") as f: with open(input_file, "w") as f:
log.debug("Writing into %s", input_file) log.debug("Writing into %s", input_file)
# fields = dataclasses.fields(PluginDesc) # fields = dataclasses.fields(PluginDesc)
fieldnames = ['repo', 'branch', 'alias'] fieldnames = ["repo", "branch", "alias"]
writer = csv.DictWriter(f, fieldnames, dialect='unix', quoting=csv.QUOTE_NONE) writer = csv.DictWriter(f, fieldnames, dialect="unix", quoting=csv.QUOTE_NONE)
writer.writeheader() writer.writeheader()
for plugin in sorted(plugins): for plugin in sorted(plugins):
writer.writerow(asdict(plugin)) writer.writerow(asdict(plugin))
@ -727,7 +757,6 @@ def commit(repo: git.Repo, message: str, files: List[Path]) -> None:
print("no changes in working tree to commit") print("no changes in working tree to commit")
def update_plugins(editor: Editor, args): def update_plugins(editor: Editor, args):
"""The main entry function of this module. All input arguments are grouped in the `Editor`.""" """The main entry function of this module. All input arguments are grouped in the `Editor`."""
@ -752,4 +781,3 @@ def update_plugins(editor: Editor, args):
f"{editor.attr_path}: resolve github repository redirects", f"{editor.attr_path}: resolve github repository redirects",
[args.outfile, args.input_file, editor.deprecated], [args.outfile, args.input_file, editor.deprecated],
) )