はじまり
リポジトリ増えてきたな・・・
一目で情報を見られるようにするか。
リポジトリを一目で見渡したい
自分のGitHubアカウントにあるリポジトリが増えてくると、一体どこのリポジトリが今どうなっているのか、どんなリポジトリが自分のアカウント上にあるのか、ふと気になってきました。
そこで、GitHubで管理されているリポジトリのデータを瞬時に確認するために、GitHub APIを利用して必要なデータを抽出する仕組みを作っていきたいと思います。
開発に使ったライブラリなど
ライブラリおよびツール一覧
今回使ったライブラリやツールは
- Python 3.10
- PyGithub（PythonからGitHub APIにアクセスしやすくなる。）
- Pytest（Pythonコードをテストする。）
- Pytest Coverage Comment（PytestのカバレッジをREADME.mdに表示するためのGitHub Actions）
- Cloud Shell Editor (Google Cloudで無料で使えるVSCodeライクのエディタ。Eclipse Theiaというエディタがベースらしい。週で利用可能時間が決まっている。)
今回、主に使用するPythonのライブラリはPyGithubです。PyGithubは、GitHub REST APIv3を簡単に利用できるようにするライブラリで、GitHubのリポジトリやユーザ情報を簡単に取得できます。
pipで当時の最新バージョンをインストールします。
pip install PyGithub>=1.55しかし、Cloud Shell上でpip installしたらエラーになりました・・・詳細は下記で。
Cloud Shell Editorについて
今回、Google Cloudから提供されている「Cloud Shell」で利用できる「Cloud Shell Editor」を利用していきたいと思います。
このエディタに関しては、解説記事を以下のページで書いています。
Cloud Shell Editorを使う場合、pipのバージョンがヤヴァい
そんなCloud Shellを触り始める時、pip installを実行するとこんなエラーメッセージが出てきます。
pipのバージョンが古いらしい。
DEPRECATION: Python 2.7 reached the end of its life on January 1st, 2020. Please upgrade your Python as Python 2.7 is no longer maintained. pip 21.0 will drop support for Python 2.7 in January 2021. More details about Python 2 support in pip can be found at https://pip.pypa.io/en/latest/development/release-process/#python-2-support pip 21.0 will remove support for this functionality.Python 2だと・・・！？
なんか知りませんが、Python 2でも利用できるように、pipのバージョンが古いんですかね・・・？
というわけで、Cloud ShellからPythonのパッケージを更新したい場合はpipをアップグレードしましょう。
sudo python3 -m pip install --upgrade pipこれで、Pygithubをインストールすることが出来ました。
pip install PyGithub>=1.55PythonでGitHubから情報を取得する
基本構造
PyGithubからGitHub APIにアクセスします。
情報を取得するPythonスクリプトの基本的な構造は以下の通りです。まず、GitHubからアクセストークンを発行して、APIにアクセスします。
from github import Github
# アクセストークンを設定
g = Github("my_access_token")
# リポジトリの情報を取得
repo = g.get_repo("my_username/my_repository")
print(repo.name)
print(repo.stargazers_count)アクセストークンのスコープは、「repo」を全チェックにしました。
リポジトリの色々な情報を取得する
たった1つのリポジトリの中にも沢山の情報が詰め込まれています。
リポジトリ名とその説明、URL、サイズ、Issueの数、フォークされた数などなど、多種多様な情報をGitHub APIから取得することが可能です。これほどの量とは、驚いた・・・
今回取得した情報は以下のような感じです。リポジトリのサイズとか、記述した言語とかは統計してみたかったのですかさず取得します。日付はISO-8601準拠にします。
def get_repo_info_in_format(repo) -> Repo_format:
"""
Retrieves and formats repository data into a dictionary conforming to Repo_format.
:param repo: The repository object to format.
:return: A dictionary containing formatted repository data.
:rtype: Repo_format
"""
state_of_pulls = "all"
created_at = repo.created_at
updated_at = repo.updated_at
created_at = f"{created_at.year:04}-{created_at.month:02}-{created_at.day:02}T{created_at.hour:02}:{created_at.minute:02}:{created_at.second:02}Z"
updated_at = f"{updated_at.year:04}-{updated_at.month:02}-{updated_at.day:02}T{updated_at.hour:02}:{updated_at.minute:02}:{updated_at.second:02}Z"
obj = {
"name": repo.name,
"description": repo.description,
"is_private": repo.private,
"html_url": repo.html_url,
"issues_count": repo.open_issues_count,
"forks_count": repo.forks_count,
"stargazers_count": repo.stargazers_count,
"subscribers_count": repo.subscribers_count,
"size": repo.size,
"is_archived": repo.archived,
"created_at": created_at,
"updated_at": updated_at,
"language": repo.language,
"languages": repo.get_languages(),
"pulls_count": repo.get_pulls(state=state_of_pulls).totalCount
}
return obj処理をマルチスレッド化する
今回取得する情報の中には、リポジトリのプルリクのカウント、また取得するプログラミング言語は複数となっています。
そのように情報を取得する場合には、GitHub APIへのリクエストは一回だけでは足りません。上記の2種類の情報を取得するために更に2回リクエストを行う必要があります。（そして、更に2回リクエストを行うリポジトリの数が100個近くあるため、200回のGETリクエストを行います・・・）
それだけリクエストの数が多いと流石に処理時間が長すぎるので、マルチスレッドにして処理していきます。
今回使ったのは、threadingというライブラリです。
import threading
from pprint import pprint
from typing import TypedDict, Final
import datetime
import json
import requests
import os
from github import Github
from config import get_config, get_github_token, get_github_username, get_env_variable
from memory_profiler import profile
def fetch_repositories(github, fetch_type: str, username: str | None):
"""
Fetches a list of repositories for a specified user or the authenticated user if no username is provided.
:param github: The GitHub session object.
:param fetch_type: The type of repositories to fetch ('all', 'owner', 'public', 'private', 'forks', etc.).
:param username: The username of the GitHub user whose repositories are to be fetched. If None, fetches repositories of the authenticated user.
:type username: str, optional
:return: A list of Repository objects.
:rtype: PaginatedList[Repository]
:raises ValueError: if fetch_type is an empty string.
"""
if fetch_type not in ["all", "owner", "public", "private", "forks"]:
raise ValueError(f"'fetch_type' is not support '{fetch_type}'")
if username == None:
user = github.get_user()
else:
user = github.get_user(username)
repos = user.get_repos(type=fetch_type)
return repos
def store_repo_info(repo, results: list) -> None:
"""
Processes and stores information about a repository in a list, intended for multi-threaded operations.
:param repo: The repository object to process information from.
:param results: The list where processed information will be stored.
"""
obj: Repo_format = get_repo_info_in_format(repo)
# gh_info.append(obj)
results.append(obj)
# @profile
def get_repo_info(github, is_threading: bool = False, username: str | None = None) -> list[Repo_format]:
"""
Starts multiple threads to process detailed information for 'owner' type repositories for a given user.
:param github: The GitHub session object.
:param username: The username of the GitHub user. If None, processes repositories for the authenticated user.
:type username: str, optional
:return: A list of processed repository information.
:rtype: list[Repo_format]
"""
repos = fetch_repositories(github, "owner", username)
results = []
if is_threading:
threads = []
for repo in repos:
thread = threading.Thread(
target=store_repo_info, args=(repo, results))
threads.append(thread)
thread.start()
pprint(threads.__sizeof__())
pprint("threads.__sizeof__()------------------------------------")
for thread in threads:
thread.join()
else:
for repo in repos:
info: Repo_format = get_repo_info_in_format(repo)
results.append(info)
return resultsこの処理の内容は、ざっとこんな感じです。
thread = threading.Thread(target=store_repo_info, args=(repo, results))で、追加するスレッドを設定する。store_repo_info()内のget_repo_info_in_format()で、更に2回リクエストを行って、リストの中に情報を入れる。thread.start()で、スレッドを追加する。thread.join()で、追加したスレッドが終了するまでget_repo_info関数が終了しないようにする。
こうすることで、200回のリクエスト処理を並行化することが出来ました。かなり処理時間を減らせました。
そしたら、出来たコードをテストします。
Pytestでテストして、カバレッジをREADME.mdに表示する
有名なリポジトリではよく、テストのカバレッジをREADMEの冒頭で表示してたりしますよね。アレを、やってみたいと思います。
今回使うのは、「Pytest Coverage Comment」および「Dynamic Badges」いうGitHub Actionsです。
「Pytest Coverage Comment」でバッジの情報を作って、「Dynamic Badges」でバッジを作ります。
今回使用したGitHub Actionsのワークフローはこちら。
name: pytest-integration
on:
push:
permissions: write-all
env:
PYTHON_VERSION: '3.10'
PYTHON_VERSION_JSON: 'python-version.json'
LICENSE: 'Apache'
LICENSE_JSON: 'license.json'
TEST_DIR: 'pytest_results'
GIST_ID: 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
PYTHON_COVERAGE_COMMENT_JSON: 'pytest-coverage-comment.json'
jobs:
test:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: ${{ env.PYTHON_VERSION }}
architecture: 'x64'
- name: Get Python version
run: python -V
- name: Install dependencies
run: pip install --no-cache-dir -r requirements/dev.txt
- name: Set environment variables
run: |
echo 'PYTHONPATH=./' >> .env
mkdir -p ${{ env.TEST_DIR }}
- name: Run pytest
run: |
python -m pytest -n auto --cov=src --cov-branch --cov-report=term-missing:skip-covered --tb=short --junitxml=./${{ env.TEST_DIR }}/junit.xml | tee ./${{ env.TEST_DIR }}/coverage.txt
- name: Create Coverage Comment
id: coverageComment
uses: MishaKav/pytest-coverage-comment@main
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
pytest-coverage-path: ./${{ env.TEST_DIR }}/coverage.txt
junitxml-path: ./${{ env.TEST_DIR }}/junit.xml
- name: Create Coverage Badge
uses: schneegans/dynamic-badges-action@v1.7.0
with:
auth: ${{ secrets.ACCESS_BADGE_IN_GIST }}
gistID: ${{ env.GIST_ID }}
filename: ${{ env.PYTHON_COVERAGE_COMMENT_JSON }}
label: Coverage
message: ${{ steps.coverageComment.outputs.coverage }}
color: ${{ steps.coverageComment.outputs.color }}
namedLogo: pytest
- name: Generate Python version json
uses: jsdaniell/create-json@v1.2.3
with:
name: ${{ env.PYTHON_VERSION_JSON }}
json: '{"label":"Python","message":"${{ env.PYTHON_VERSION }}","schemaVersion":1,"color":"blue","namedLogo":"python","style":"flat"}'
dir: './'
- name: Deploy Python version json to Gist
uses: exuanbo/actions-deploy-gist@v1.1.4
with:
token: ${{ secrets.ACCESS_BADGE_IN_GIST }}
gist_id: ${{ env.GIST_ID }}
file_path: ${{ env.PYTHON_VERSION_JSON }}
file_type: text
- name: Generate License json
uses: jsdaniell/create-json@v1.2.3
with:
name: ${{ env.LICENSE_JSON }}
json: '{"label":"license","message":"${{ env.LICENSE }}","schemaVersion":1,"color":"skyblue","namedLogo":"apache","style":"flat"}'
dir: './'
- name: Deploy License json to Gist
uses: exuanbo/actions-deploy-gist@v1.1.4
with:
token: ${{ secrets.ACCESS_BADGE_IN_GIST }}
gist_id: ${{ env.GIST_ID }}
file_path: ${{ env.LICENSE_JSON }}
file_type: textPytestのカバレッジ以外にも、Pythonのバージョンおよびライセンスの情報もバッジとして表示できるようにしています。その場合、「Pytest Coverage Comment」を使わないでJSONファイルを作る必要があるので、「jsdaniell/create-json」のGitHub Actionsを利用しています。そして、そのJSON形式の情報を「exuanbo/actions-deploy-gist」のGitHub Actionsを利用してGistにアップロードする。。。（色々なActionsを使わせてもらいました。ありがたい。）
面倒なので、pushした時に動くようにしてしまっています。
バッジがちゃんと追加できているとこんな感じ。
Cloud Functionsにデプロイする
PythonでGitHub情報を取得して、テストも行えたならば、Cloud Functionsとして本処理をデプロイしていきたいと思います。（今回取得する情報は、GAS（Google Apps Script）から取ってGoogleスプレッドシートに入れたかったのです。）
Google Cloudのコンソールからデプロイしていきたいと思います。
import functions_framework
@functions_framework.http
def retrieve_github_repo_info_by_request(request) -> str:
"""
Handles an HTTP request to retrieve GitHub repository information based on the provided token.
:param request: The HTTP request object containing parameters and JSON data.
:return: A JSON formatted string of repository information or a greeting message if no token is provided.
:rtype: str
"""
request_json = request.get_json(silent=True)
request_args = request.args
info: list[Repo_format] | str = []
if request_json and "token" in request_json:
token: str = request_json["token"]
info = retrieve_github_repo_info(token, False)
elif request_args and "token" in request_args:
info = request_args["token"]
else:
info = "Hello, World!!!"
res_obj = {"data": info}
return json.dumps(res_obj, sort_keys=True, ensure_ascii=False)Google Cloudでは、Pythonランタイムを使用する場合に、コンソール画面でデプロイ直前の事前テストを行うことが出来ます。
便利なのですが、環境変数を使ったテストは出来なかったりします。その部分は、別記事でまとめています。
デプロイ後の格闘
デプロイ後に直面するのが、APIの様々な閾値に起因するバグです。
GASからリクエストをした時に、こんなエラーメッセージが返ってきたりしました。
Service Unavailable
{"object":"error","status":503,"code":"service_unavailable","message":"Public API service is temporarily unavailable, please try again later.","request_id":"xxxxxxxc-xxxx-4d07-bcc3-xxxxxxxxx"}このメッセージはレスポンスコード503なので、Cloud Functions上でバグっているみたいです。
最初はタイムアウトすることが原因だったのですが、何回かリクエストしていると、メモリ不足で起きるエラーも発生していました。
なので、デプロイした関数に割り当てるメモリ量やタイムアウト閾値を増やします。
一旦、これで動くようになりました！
Googleスプレッドシートに入れたらこんな感じ。
これで、PythonでGitHub上のリポジトリの情報を取得するCloud Functionsが出来ました！
おっ・・・？
まとめ
この記事では、GitHubのリポジトリ情報をPythonとCloud Functionsを使用して取得する手順を解説しました。
- GitHub APIを利用してリポジトリ情報を抽出しました。
- ライブラリとツール：PyGithub、Pytest、Cloud Shell Editorなど。
- マルチスレッド処理：情報取得処理を効率化しました。
- Pytestでのテストおよびカバレッジの表示：GitHub Actionsを活用して自動化しました。
- Cloud Functionsによるデプロイ：Googleスプレッドシートに保存できるようにAPI化しました。
GASだとAPIを叩きやすくするライブラリがあまり無いので、今回はPythonで実装しました。
Cloud Functionsは。1日に1リクエストくらいであれば月に5円も掛からないので、これからも利用していきたいです。
おしまい
なんかメモリ食いすぎじゃね？
うーん確かに・・・
以上になります！
コメント