|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +import datetime as dt |
| 4 | +import operator |
| 5 | +import os |
| 6 | +from dataclasses import dataclass |
| 7 | + |
| 8 | +from gql import Client, gql |
| 9 | +from gql.transport.httpx import HTTPXTransport |
| 10 | +from graphql import DocumentNode |
| 11 | +from httpx import Timeout |
| 12 | + |
| 13 | +TOK = os.environ.get("PUBLIC_PAT", "") |
| 14 | +if not TOK: |
| 15 | + raise RuntimeError("A PAT could not be found in the environment, please set 'PUBLIC_PAT'") |
| 16 | + |
| 17 | + |
| 18 | +# This search query should yield all of my public repositories, sorted by last updated |
| 19 | +# For each repository, provide: |
| 20 | +# * Name |
| 21 | +# * URL |
| 22 | +# * Is archived? |
| 23 | +# * Releases (may be empty) |
| 24 | +# * Last release |
| 25 | +# * Tag |
| 26 | +# * Publish date |
| 27 | +# * URL |
| 28 | +# * Total number of releases |
| 29 | +# This query is limited to 50 results at a time, so pagination information is also provided |
| 30 | +BASE_QUERY = """ |
| 31 | +query { |
| 32 | + search( |
| 33 | + first: 50 |
| 34 | + type: REPOSITORY |
| 35 | + query: "owner:sco1 is:public sort:updated" |
| 36 | + after: %AFTER% |
| 37 | + ) { |
| 38 | + pageInfo { |
| 39 | + hasNextPage |
| 40 | + endCursor |
| 41 | + } |
| 42 | + nodes { |
| 43 | + ... on Repository { |
| 44 | + name |
| 45 | + url |
| 46 | + isArchived |
| 47 | +
|
| 48 | + releases(orderBy: { field: CREATED_AT, direction: DESC }, first: 1) { |
| 49 | + totalCount |
| 50 | + nodes { |
| 51 | + tagName |
| 52 | + publishedAt |
| 53 | + url |
| 54 | + } |
| 55 | + } |
| 56 | + } |
| 57 | + } |
| 58 | + } |
| 59 | +} |
| 60 | +""" |
| 61 | + |
| 62 | +TIMEOUT = Timeout(5, read=10) |
| 63 | +TRANSPORT = HTTPXTransport( |
| 64 | + url="https://api.github.com/graphql", |
| 65 | + headers={"Authorization": f"bearer {TOK}"}, |
| 66 | + timeout=TIMEOUT, |
| 67 | +) |
| 68 | +CLIENT = Client(transport=TRANSPORT, fetch_schema_from_transport=True) |
| 69 | + |
| 70 | + |
| 71 | +@dataclass(slots=True, frozen=True) |
| 72 | +class Release: # noqa: D101 |
| 73 | + tag_name: str |
| 74 | + published: dt.datetime |
| 75 | + url: str |
| 76 | + |
| 77 | + @classmethod |
| 78 | + def from_node(cls, node: dict) -> Release: |
| 79 | + """ |
| 80 | + Build a `Release` instance from the provided node. |
| 81 | +
|
| 82 | + The node is assumed to contain the following keys: |
| 83 | + * `"tagName"` |
| 84 | + * `"publishedAt"` |
| 85 | + * `"url"` |
| 86 | + """ |
| 87 | + return cls( |
| 88 | + tag_name=node["tagName"], |
| 89 | + published=dt.datetime.fromisoformat(node["publishedAt"]), |
| 90 | + url=node["url"], |
| 91 | + ) |
| 92 | + |
| 93 | + |
| 94 | +@dataclass(slots=True, frozen=True) |
| 95 | +class Repository: # noqa: D101 |
| 96 | + name: str |
| 97 | + url: str |
| 98 | + n_releases: int |
| 99 | + last_release: Release |
| 100 | + |
| 101 | + @classmethod |
| 102 | + def from_node(cls, node: dict) -> Repository: |
| 103 | + """ |
| 104 | + Build a `Repository` instance from the provided node. |
| 105 | +
|
| 106 | + The node is assumed to contain the following keys: |
| 107 | + * `"name"` |
| 108 | + * `"url"` |
| 109 | + * `"releases"` (of a form expected by `Release.from_node`) |
| 110 | + """ |
| 111 | + return cls( |
| 112 | + name=node["name"], |
| 113 | + url=node["url"], |
| 114 | + n_releases=node["releases"]["totalCount"], |
| 115 | + last_release=Release.from_node(node["releases"]["nodes"][0]), |
| 116 | + ) |
| 117 | + |
| 118 | + |
| 119 | +def _paginate_query(after: str | None = None) -> DocumentNode: |
| 120 | + """ |
| 121 | + Update `BASE_QUERY` with the provided pagination cursor. |
| 122 | +
|
| 123 | + Alternatively, if `None` is passed, the cursor is specified as `"null"` to obtain the first page |
| 124 | + of results. |
| 125 | +
|
| 126 | + NOTE: It is assumed that `BASE_QUERY` contains an `%AFTER%` placeholder indicating where the |
| 127 | + cursor should be inserted. |
| 128 | + """ |
| 129 | + if after is None: |
| 130 | + insert = "null" |
| 131 | + else: |
| 132 | + # If a cursor is provided it needs to be wrapped in quotes |
| 133 | + insert = f'"{after}"' |
| 134 | + |
| 135 | + query = BASE_QUERY.replace("%AFTER%", insert) |
| 136 | + return gql(query) |
| 137 | + |
| 138 | + |
| 139 | +def n_recent_releases(n: int = 5) -> list[Release]: |
| 140 | + """ |
| 141 | + Query the GitHub GraphQL API for my `n` most recent repositories with a published release. |
| 142 | +
|
| 143 | + For the purposes of this tool, only public, non-archived repositories are considered. |
| 144 | + """ |
| 145 | + repositories = [] |
| 146 | + |
| 147 | + has_page = True |
| 148 | + after: str | None = None |
| 149 | + while has_page: |
| 150 | + query = _paginate_query(after=after) |
| 151 | + result = CLIENT.execute(query) |
| 152 | + |
| 153 | + has_page = result["search"]["pageInfo"]["hasNextPage"] |
| 154 | + after = result["search"]["pageInfo"]["endCursor"] |
| 155 | + |
| 156 | + for repo in result["search"]["nodes"]: |
| 157 | + if repo["isArchived"]: |
| 158 | + continue |
| 159 | + |
| 160 | + release_nodes = repo["releases"]["nodes"] |
| 161 | + if not release_nodes: |
| 162 | + continue |
| 163 | + |
| 164 | + repositories.append(Repository.from_node(repo)) |
| 165 | + |
| 166 | + repositories.sort(key=operator.attrgetter("last_release.published"), reverse=True) |
| 167 | + return repositories[:n] |
| 168 | + |
| 169 | + |
| 170 | +if __name__ == "__main__": |
| 171 | + n_recent_releases() |
0 commit comments