Fix deletion of assets from nightly release on GitHub

The derived asset URL was wrong. Instead use the URL supplied in the
JSON asset description.
This commit is contained in:
Kovid Goyal 2023-03-02 11:28:05 +05:30
parent 294d36f2d3
commit 719fe9ea04
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C

View File

@ -20,7 +20,7 @@ import tempfile
import time import time
from contextlib import contextmanager, suppress from contextlib import contextmanager, suppress
from http.client import HTTPResponse, HTTPSConnection from http.client import HTTPResponse, HTTPSConnection
from typing import Any, Callable, Dict, Generator, Iterable, Optional, Tuple, Union from typing import Any, Callable, Dict, Generator, Iterable, List, Optional, Tuple, Union
from urllib.parse import urlencode, urlparse from urllib.parse import urlencode, urlparse
os.chdir(os.path.dirname(os.path.abspath(__file__))) os.chdir(os.path.dirname(os.path.abspath(__file__)))
@ -325,66 +325,44 @@ class GitHub: # {{{
' For how to install nightly builds, see: https://sw.kovidgoyal.net/kitty/binary/#customizing-the-installation' ' For how to install nightly builds, see: https://sw.kovidgoyal.net/kitty/binary/#customizing-the-installation'
) )
def delete_asset(self, url: str, fname: str) -> None:
self.make_request_with_retries(
url, method='DELETE', num_tries=5, sleep_between_tries=2,
success_codes=(204, 404),
failure_msg=f'Failed to delete {fname} from GitHub')
def __call__(self) -> None: def __call__(self) -> None:
# See https://docs.github.com/en/rest/releases/assets#upload-a-release-asset # See https://docs.github.com/en/rest/releases/assets#upload-a-release-asset
# self.clean_older_releases(releases)
release = self.create_release() release = self.create_release()
upload_url = release['upload_url'].partition('{')[0] upload_url = release['upload_url'].partition('{')[0]
existing_assets = self.existing_assets(release) all_assest_for_release = self.existing_assets_for_release(release)
assets_by_fname = {a['name']:a for a in all_assest_for_release}
def delete_asset(asset_id: str) -> None: def delete_asset(asset: Dict[str, Any], allow_not_found: bool = True) -> None:
asset_url = release['assets_url'] + f'/{asset_id}' success_codes = [204]
self.delete_asset(asset_url, fname) if allow_not_found:
success_codes.append(404)
self.make_request_with_retries(
asset['url'], method='DELETE', num_tries=5, sleep_between_tries=2, success_codes=tuple(success_codes),
failure_msg=f'Failed to delete {fname} from GitHub')
def upload_with_retries(path: str, desc: str, num_tries: int = 8, sleep_time: float = 60.0) -> None: def upload_with_retries(path: str, desc: str, num_tries: int = 8, sleep_time: float = 60.0) -> None:
fname = os.path.basename(path) fname = os.path.basename(path)
if self.is_nightly: if self.is_nightly:
fname = fname.replace(version, 'nightly') fname = fname.replace(version, 'nightly')
if fname in existing_assets: if fname in assets_by_fname:
self.info(f'Deleting {fname} from GitHub with id: {existing_assets[fname]}') self.info(f'Deleting {fname} from GitHub with id: {assets_by_fname[fname]["id"]}')
delete_asset(existing_assets.pop(fname)) delete_asset(assets_by_fname.pop(fname))
params = {'name': fname, 'label': desc} params = {'name': fname, 'label': desc}
def handle_failure(r: HTTPResponse) -> None:
try:
asset_id = json.loads(r.read())['id']
except Exception:
try:
asset_id = self.existing_assets(release['id'])[fname]
except KeyError:
asset_id = 0
if asset_id:
self.info(f'Deleting {fname} from GitHub with id: {asset_id}')
delete_asset(asset_id)
self.make_request_with_retries( self.make_request_with_retries(
upload_url, upload_path=path, params=params, num_tries=num_tries, sleep_between_tries=sleep_time, upload_url, upload_path=path, params=params, num_tries=num_tries, sleep_between_tries=sleep_time,
failure_msg=f'Failed to upload file: {fname}', success_codes=(201,), failure_callback=handle_failure failure_msg=f'Failed to upload file: {fname}', success_codes=(201,),
) )
if self.is_nightly: if self.is_nightly:
for fname in tuple(existing_assets): for fname in tuple(assets_by_fname):
self.info(f'Deleting {fname} from GitHub with id: {existing_assets[fname]}') self.info(f'Deleting {fname} from GitHub with id: {assets_by_fname[fname]["id"]}')
delete_asset(existing_assets.pop(fname)) delete_asset(assets_by_fname.pop(fname), allow_not_found=False)
self.update_nightly_description(release['id'])
for path, desc in self.files.items(): for path, desc in self.files.items():
self.info('') self.info('')
upload_with_retries(path, desc) upload_with_retries(path, desc)
if self.is_nightly:
def clean_older_releases(self, releases: Iterable[Dict[str, Any]]) -> None: self.update_nightly_description(release['id'])
for release in releases:
if release.get('assets') and release['tag_name'] != self.current_tag_name:
self.info(f'\nDeleting old released installers from: {release["tag_name"]}')
for asset in release['assets']:
self.delete_asset(
f'{self.url_base}/assets/{asset["id"]}', asset['name'])
def print_failed_response_details(self, r: HTTPResponse, msg: str) -> None: def print_failed_response_details(self, r: HTTPResponse, msg: str) -> None:
self.error(msg, f'\nStatus Code: {r.status} {r.reason}') self.error(msg, f'\nStatus Code: {r.status} {r.reason}')
@ -400,13 +378,13 @@ class GitHub: # {{{
self.print_failed_response_details(r, msg) self.print_failed_response_details(r, msg)
raise SystemExit(1) raise SystemExit(1)
def existing_assets(self, release: Dict[str, Any]) -> Dict[str, str]: def existing_assets_for_release(self, release: Dict[str, Any]) -> List[Dict[str, Any]]:
if 'assets' in release: if 'assets' in release:
d = release['assets'] d: List[Dict[str, Any]] = release['assets']
else: else:
d = self.make_request_with_retries( d = self.make_request_with_retries(
release['assets_url'], params={'per_page': '64'}, failure_msg='Failed to get assets for release', return_data=True) release['assets_url'], params={'per_page': '64'}, failure_msg='Failed to get assets for release', return_data=True)
return {asset['name']: asset['id'] for asset in d} return d
def create_release(self) -> Dict[str, Any]: def create_release(self) -> Dict[str, Any]:
' Create a release on GitHub or if it already exists, return the existing release ' ' Create a release on GitHub or if it already exists, return the existing release '