dotfiles

 1#!/usr/bin/env python3
 2
 3import os
 4import sys
 5import json
 6import asyncio
 7import aiohttp
 8import argparse
 9import subprocess
10from typing import List
11from dataclasses import dataclass
12
13
14try:
15        _github_token = subprocess.check_output(['pass', 'dev/github.com/repo_token']).decode().strip()
16except:
17        _github_token = ""
18
19outlocker = asyncio.Lock()
20headers={
21        "X-GitHub-Api-Version": "2022-11-28",
22        "Accept": "application/vnd.github+json",
23        "Authorization": _github_token,
24}
25
26@dataclass
27class RepoInfo:
28        name: str
29        version: str
30        commit: str = "unknown"
31
32async def list_tags(sess: aiohttp.ClientSession, repo: str, num: int = 3) -> List[RepoInfo]:
33        info = list()
34        resp = await sess.get(f"/repos/{repo}/tags", params={"per_page": num})
35        assert resp.status == 200, await resp.text()
36        data = await resp.json()
37        async with outlocker:
38                for i in range(min(num, len(data))):
39                        info.append(RepoInfo(
40                                name = repo,
41                                version=data[i].get("name"),
42                                commit = data[i].get("commit", {}).get("sha")
43                        ))
44        return info
45
46async def list_releases(sess: aiohttp.ClientSession, repo: str, num: int=3) -> List[RepoInfo]:
47        info = list()
48        resp = await sess.get(f"/repos/{repo}/releases", params={"per_page": num})
49        assert resp.status == 200, await resp.text()
50        data = await resp.json()
51        async with outlocker:
52                for i in range(min(num, len(data))):
53                        info.append(RepoInfo(
54                                name = repo,
55                                version = data[i].get("tag_name")
56                        ))
57                        #print(f'{repo: <20} {data[i].get("name"): <30} {data[i].get("tag_name", "None")}')
58        return info
59        
60async def main():
61    parser = argparse.ArgumentParser()
62    
63    subparsers = parser.add_subparsers(dest="cmd")
64    tags_cmd = subparsers.add_parser("tags")
65    tags_cmd.add_argument("-n", "--num", default=3, type=int)
66    tags_cmd.add_argument("repo", nargs='+')
67    
68    releases_cmd = subparsers.add_parser("rels")
69    releases_cmd.add_argument("-n", "--num", default=3, type=int)
70    releases_cmd.add_argument("repo", nargs='+')
71
72    args = parser.parse_args()
73
74    async with aiohttp.ClientSession(base_url="https://api.github.com", headers=headers) as sess:
75        if args.cmd == "tags":
76                infos = await asyncio.gather(*[list_tags(sess, repo, args.num) for repo in args.repo])
77        elif args.cmd == "rels":
78                infos = await asyncio.gather(*[list_releases(sess, repo, args.num) for repo in args.repo])
79        for line in infos:
80                for info in line:
81                        print(f"{info.name: <20}\t{info.version: <30}\t{info.commit}")
82        
83        
84if __name__ == '__main__':    
85        asyncio.run(main())