Formatting

package/scripts/publish_latest_generic_package.py

@@ -13,7 +13,6 @@ DEFAULT_REGISTRY_URL = "https://git.thyme-labs.com/api/packages/khodo/generic"
 DEFAULT_VERSION = "latest"
 PAGE_LIMIT = 50
 
-
 def read_config(path):
   try:
     data = json.loads(path.read_text(encoding="utf-8"))
@@ -33,7 +32,6 @@ def read_config(path):
     registry_url = DEFAULT_REGISTRY_URL
   return name, registry_url
 
-
 def parse_registry_url(registry_url):
   parsed = urllib.parse.urlparse(registry_url.rstrip("/"))
   parts = [urllib.parse.unquote(part) for part in parsed.path.split("/") if part]
@@ -43,7 +41,6 @@ def parse_registry_url(registry_url):
   prefix = "/" + "/".join(urllib.parse.quote(part, safe="") for part in parts[:-4]) if parts[:-4] else ""
   return parsed, prefix, owner
 
-
 def build_url(parsed, path, query=""):
   return urllib.parse.urlunparse((
     parsed.scheme,
@@ -54,11 +51,9 @@ def build_url(parsed, path, query=""):
     "",
   ))
 
-
 def quote_path(*parts):
   return "/".join(urllib.parse.quote(part, safe="") for part in parts)
 
-
 def version_list_url(parsed, prefix, owner, package_name, page):
   path = f"{prefix}/api/v1/packages/{quote_path(owner, 'generic', package_name)}"
   query = urllib.parse.urlencode({
@@ -67,12 +62,10 @@ def version_list_url(parsed, prefix, owner, package_name, page):
   })
   return build_url(parsed, path, query)
 
-
 def version_delete_url(parsed, prefix, owner, package_name, version):
   path = f"{prefix}/api/packages/{quote_path(owner, 'generic', package_name, version)}"
   return build_url(parsed, path)
 
-
 def file_upload_url(registry_url, package_name, version, path):
   return "/".join((
     registry_url.rstrip("/"),
@@ -81,12 +74,10 @@ def file_upload_url(registry_url, package_name, version, path):
     urllib.parse.quote(path.name, safe=""),
   ))
 
-
 def authorization(user, token):
   encoded = base64.b64encode(f"{user}:{token}".encode("utf-8")).decode("ascii")
   return f"Basic {encoded}"
 
-
 def request(method, url, auth, data=None, headers=None, missing_ok=False):
   request_headers = {
     "Authorization": auth,
@@ -106,7 +97,6 @@ def request(method, url, auth, data=None, headers=None, missing_ok=False):
   except urllib.error.URLError as error:
     raise SystemExit(f"{method} {url} failed: {error.reason}") from error
 
-
 def list_versions(parsed, prefix, owner, package_name, auth):
   versions = []
   page = 1
@@ -128,7 +118,6 @@ def list_versions(parsed, prefix, owner, package_name, auth):
     page += 1
   return sorted(set(versions))
 
-
 def delete_version(parsed, prefix, owner, package_name, version, auth):
   url = version_delete_url(parsed, prefix, owner, package_name, version)
   response = request("DELETE", url, auth, missing_ok=True)
@@ -140,7 +129,6 @@ def delete_version(parsed, prefix, owner, package_name, version, auth):
       raise SystemExit(f"DELETE {url} returned HTTP {response.status}")
   print(f"Deleted package version: {package_name}/{version}")
 
-
 def discover_artifacts(dist_dir, package_name, version):
   archives = sorted(dist_dir.glob(f"{package_name}-{version}-*.tar.gz"))
   artifacts = []
@@ -156,7 +144,6 @@ def discover_artifacts(dist_dir, package_name, version):
     raise SystemExit("missing artifact files: " + ", ".join(str(path) for path in missing))
   return artifacts
 
-
 def upload_file(registry_url, package_name, version, path, auth):
   url = file_upload_url(registry_url, package_name, version, path)
   headers = {
@@ -170,7 +157,6 @@ def upload_file(registry_url, package_name, version, path, auth):
         raise SystemExit(f"PUT {url} returned HTTP {response.status}")
   print(f"Uploaded package file: {package_name}/{version}/{path.name}")
 
-
 def parse_args(argv):
   parser = argparse.ArgumentParser(description="Replace Product Vamp's rolling Gitea generic package.")
   parser.add_argument("--config", type=pathlib.Path, default=pathlib.Path("modulus.json"))
@@ -180,7 +166,6 @@ def parse_args(argv):
   parser.add_argument("--token", default=None)
   return parser.parse_args(argv)
 
-
 def main(argv):
   args = parse_args(argv)
   package_name, registry_url = read_config(args.config)
@@ -189,30 +174,23 @@ def main(argv):
   auth = authorization(user, token)
   parsed, prefix, owner = parse_registry_url(registry_url)
   artifacts = discover_artifacts(args.dist_dir, package_name, args.version)
-
   existing_versions = list_versions(parsed, prefix, owner, package_name, auth)
-
   # Gitea rejects duplicate file names within a generic package version, so
   # replacing the rolling release requires deleting the old version first.
   if args.version in existing_versions:
     delete_version(parsed, prefix, owner, package_name, args.version, auth)
-
   for artifact in artifacts:
     upload_file(registry_url, package_name, args.version, artifact, auth)
-
   for version in existing_versions:
     if version != args.version:
       delete_version(parsed, prefix, owner, package_name, version, auth)
-
   return 0
 
-
 def os_environ(name):
   value = os.environ.get(name)
   if not value:
     raise SystemExit(f"{name} is required")
   return value
 
-
 if __name__ == "__main__":
   raise SystemExit(main(sys.argv[1:]))