All defs are now typed

This commit is contained in:
Kovid Goyal
2020-03-14 14:37:11 +05:30
parent d3f37eeba4
commit 5956277863
5 changed files with 272 additions and 191 deletions

View File

@@ -16,7 +16,7 @@ import sys
import tempfile
import time
from contextlib import suppress
from typing import IO, Optional, cast
from typing import IO, Any, Dict, Iterable, List, Optional, cast
import requests
@@ -44,7 +44,7 @@ def call(*cmd: str, cwd: Optional[str] = None) -> None:
raise SystemExit(ret)
def run_build(args):
def run_build(args: Any) -> None:
os.chdir(build_path)
call('./linux 64 kitty')
call('./osx kitty --sign-installers')
@@ -52,21 +52,21 @@ def run_build(args):
call('./linux 32 kitty')
def run_tag(args):
def run_tag(args: Any) -> None:
call('git push')
call('git tag -s v{0} -m version-{0}'.format(version))
call('git push origin v{0}'.format(version))
def run_man(args):
def run_man(args: Any) -> None:
call('make FAIL_WARN=-W man', cwd=docs_dir)
def run_html(args):
def run_html(args: Any) -> None:
call('make FAIL_WARN=-W html', cwd=docs_dir)
def add_analytics():
def add_analytics() -> None:
analytics = '''
<!-- Google Analytics -->
<script>
@@ -87,7 +87,7 @@ ga('send', 'pageview');
f.write(html.encode('utf-8'))
def run_website(args):
def run_website(args: Any) -> None:
if os.path.exists(publish_dir):
shutil.rmtree(publish_dir)
shutil.copytree(os.path.join(docs_dir, '_build', 'html'), publish_dir)
@@ -105,7 +105,7 @@ def run_website(args):
subprocess.check_call(['git', 'push'])
def run_sdist(args):
def run_sdist(args: Any) -> None:
with tempfile.TemporaryDirectory() as tdir:
base = os.path.join(tdir, f'kitty-{version}')
os.mkdir(base)
@@ -122,7 +122,7 @@ def run_sdist(args):
class ReadFileWithProgressReporting(io.FileIO): # {{{
def __init__(self, path):
def __init__(self, path: str):
io.FileIO.__init__(self, path, 'rb')
self.seek(0, os.SEEK_END)
self._total = self.tell()
@@ -166,14 +166,12 @@ class ReadFileWithProgressReporting(io.FileIO): # {{{
class Base(object): # {{{
def __init__(self):
pass
def info(self, *args, **kwargs):
def info(self, *args: Any, **kwargs: Any) -> None:
print(*args, **kwargs)
sys.stdout.flush()
def warn(self, *args, **kwargs):
def warn(self, *args: Any, **kwargs: Any) -> None:
print('\n' + '_' * 20, 'WARNING', '_' * 20)
print(*args, **kwargs)
print('_' * 50)
@@ -187,13 +185,15 @@ class GitHub(Base): # {{{
API = 'https://api.github.com/'
def __init__(self,
files,
reponame,
version,
username,
password,
replace=False):
def __init__(
self,
files: Dict[str, str],
reponame: str,
version: str,
username: str,
password: str,
replace: bool = False
):
self.files, self.reponame, self.version, self.username, self.password, self.replace = (
files, reponame, version, username, password, replace)
self.current_tag_name = 'v' + self.version
@@ -201,7 +201,7 @@ class GitHub(Base): # {{{
s.auth = (self.username, self.password)
s.headers.update({'Accept': 'application/vnd.github.v3+json'})
def __call__(self):
def __call__(self) -> None:
releases = self.releases()
# self.clean_older_releases(releases)
release = self.create_release(releases)
@@ -239,7 +239,7 @@ class GitHub(Base): # {{{
if r.status_code != 200:
self.fail(r, 'Failed to set label for %s' % fname)
def clean_older_releases(self, releases):
def clean_older_releases(self, releases: Iterable[Dict[str, Any]]) -> None:
for release in releases:
if release.get(
'assets',
@@ -256,7 +256,7 @@ class GitHub(Base): # {{{
'Failed to delete obsolete asset: %s for release: %s'
% (asset['name'], release['tag_name']))
def do_upload(self, url, path, desc, fname):
def do_upload(self, url: str, path: str, desc: str, fname: str) -> requests.Response:
mime_type = mimetypes.guess_type(fname)[0]
self.info('Uploading to GitHub: %s (%s)' % (fname, mime_type))
with ReadFileWithProgressReporting(path) as f:
@@ -269,17 +269,17 @@ class GitHub(Base): # {{{
params={'name': fname},
data=cast(IO[bytes], f))
def fail(self, r, msg):
def fail(self, r: requests.Response, msg: str) -> None:
print(msg, ' Status Code: %s' % r.status_code, file=sys.stderr)
print("JSON from response:", file=sys.stderr)
pprint.pprint(dict(r.json()), stream=sys.stderr)
raise SystemExit(1)
def already_exists(self, r):
def already_exists(self, r: requests.Response) -> bool:
error_code = r.json().get('errors', [{}])[0].get('code', None)
return error_code == 'already_exists'
return bool(error_code == 'already_exists')
def existing_assets(self, release_id):
def existing_assets(self, release_id: str) -> Dict[str, str]:
url = self.API + 'repos/%s/%s/releases/%s/assets' % (
self.username, self.reponame, release_id)
r = self.requests.get(url)
@@ -287,22 +287,21 @@ class GitHub(Base): # {{{
self.fail(r, 'Failed to get assets for release')
return {asset['name']: asset['id'] for asset in r.json()}
def releases(self):
def releases(self) -> List[Dict[str, Any]]:
url = self.API + 'repos/%s/%s/releases' % (self.username, self.reponame
)
r = self.requests.get(url)
if r.status_code != 200:
self.fail(r, 'Failed to list releases')
return r.json()
return list(r.json())
def create_release(self, releases):
def create_release(self, releases: Iterable[Dict[str, str]]) -> Dict[str, Any]:
' Create a release on GitHub or if it already exists, return the existing release '
for release in releases:
# Check for existing release
if release['tag_name'] == self.current_tag_name:
return release
url = self.API + 'repos/%s/%s/releases' % (self.username, self.reponame
)
url = self.API + 'repos/%s/%s/releases' % (self.username, self.reponame)
r = self.requests.post(
url,
data=json.dumps({
@@ -316,19 +315,17 @@ class GitHub(Base): # {{{
if r.status_code != 201:
self.fail(r, 'Failed to create release for version: %s' %
self.version)
return r.json()
return dict(r.json())
# }}}
def get_github_data():
def get_github_data() -> Dict[str, str]:
with open(os.environ['PENV'] + '/github-token') as f:
un, pw = f.read().strip().split(':')
return {'username': un, 'password': pw}
def run_upload(args):
def run_upload(args: Any) -> None:
files = {
os.path.join(build_path, 'build', f.format(version)): desc
for f, desc in {
@@ -346,18 +343,18 @@ def run_upload(args):
gh()
def require_git_master(branch='master'):
def require_git_master(branch: str = 'master') -> None:
b = subprocess.check_output(['git', 'symbolic-ref', '--short', 'HEAD']).decode('utf-8').strip()
if b != branch:
raise SystemExit('You must be in the {} git branch'.format(branch))
def require_penv():
def require_penv() -> None:
if 'PENV' not in os.environ:
raise SystemExit('The PENV env var is not present, required for uploading releases')
def main():
def main() -> None:
require_git_master()
require_penv()
parser = argparse.ArgumentParser(description='Publish kitty')