G-gen の佐伯です。前編では PyGithub の管理タスク自動化の事前準備・導入について紹介しました。後編では新規ブランチの作成、ブランチ内ファイルの更新・削除、プルリクエストの自動化についてご紹介したいと思います。

はじめに

前編記事

当記事では、PyGithub の利用方法を解説しています。PyGithub は、GitHub の API を Python から利用するためのライブラリです。前提となる利用方法は前編でも解説していますので、ご参照ください。

blog.g-gen.co.jp

当記事でやること

当記事（後編）では、新規ブランチの作成、ブランチ内ファイルの更新・削除、プルリクエストの自動化として、以下の①〜④のタスクを実行します。

新規作成ブランチ（名称: stg_dev）を作成
stg_dev ブランチ内にファイル update_contents.txt を新規作成（BigQuery から読み取ったデータを書き込み）
既存ファイル test.txt を削除
ブランチ main へのプルリクエスト

事前準備

データを準備

今回は、BigQuery に以下のようなデータを準備し、これらのカラムを update_contents.txt に書き込むことにします。

{
  'id': 1, 
  'group_email': 'google-1234@g-gen.co.jp', 
  'subnet_info': '10.0.25.0/24', 
  'UPDATE_FLG': 'DLT', 
  'client_cidr': '123.0.10.12/24',
  'use_purpose': 'normal'
}

Github リポジトリを作成・main ブランチにファイルを準備

今回はGithubに以下のようなリポジトリを作成し、main ブランチに test.txt ファイルを作成します。リポジトリ名は Osamu-Saiki/test です。

Github のシークレットアカウントの作成や PyGithub のインストール等基本的な操作手法については、前編を確認ください。

新規ブランチ作成

sha に main ブランチを指定し、新規ブランチを create_git_ref メソッドで作成します。

# 新しいブランチを作成
repo.create_git_ref(ref=f'refs/heads/{new_branch_name}', sha=repo.get_branch(main_branch).commit.sha)

sha (Secure Hash Algorithm) はセキュアなハッシュ関数の一種で、データの一意な識別子を生成するために使用されます。

ファイルの新規作成

create_fileメソッドに、ファイルパス・メッセージ・更新内容・更新ブランチをそれぞれ設定して実行します。

repo.create_file(
  #ファイルパス
  path=updatefile, 
  #メッセージ
  message="update_contents.txtファイルを作成", 
  #更新内容(string)
  content=update_data,
  #更新ブランチ
  branch=new_branch_name 
)

ファイルの更新

create_file メソッドに、ファイルパス・メッセージ・更新内容・更新ブランチ・sha をそれぞれ設定して実行します。

repo.update_file(
  # 転送元のブランチのファイルパス指定
  path=updatefile,  
  #メッセージ
  message=f"{updatefile}ファイルを更新",
  #更新内容(string)
  content=update_data,
  # 送信先のブランチ指定 
  branch=new_branch_name,  
  #shaの指定
  sha=update_file_class.sha 
)

ファイルの削除

delete_file メソッドに、ファイルパス・メッセージ・sha・対象ブランチをそれぞれ設定します。

repo.delete_file(
  #ファイルパス
  path=deletefile,
  #メッセージ
  message=f"{deletefile}を削除",
  #shaの指定
  sha=delete_contents.sha,
  #対象ブランチ
  branch=new_branch_name
)

プルリクエスト

create_pull メソッドにプルリクのタイトル・内容・プルリク元のブランチ・マージする先のブランチをそれぞれ設定します。

pull_req = repo.create_pull(
  # プルリクエストのタイトル
  title="modify repository",
  # プルリクエストの内容
  body=f"Add updated {updatefile} file",
  # プルリク元のブランチを指定
  head=new_branch_name,  
  # プルリクエストをマージする先のブランチを指定
  base=main_branch  
)

プログラムの実行

ソースコード

以下のようなソースコードを実行することで stg_dev ブランチ（新規ブランチ）が Osamu-Saiki/test リポジトリに作成され main ブランチにプルリクが出されます。

①config_data.json

{
  "LOG_LEVEL" : 20,
  "access_token" : "ghp_*************************",
  "repository_name" : "Osamu-Saiki/test",
  "main_branch" : "main",
  "credentials_file" : "/home/saikio/.ssh/saikio-************.json",
  "project" : "saikio",
  "dataset_id" : "user_regist",
  "table_id" : "user_regist_list"
}

②ソースコード

from github import Github, GithubException
from google.cloud import bigquery
from google.oauth2 import service_account
  
import traceback
import logging  
import json
    
with open('config_data.json','r') as f:
    config = json.load(f)  
  
LOG_LEVEL = config["LOG_LEVEL"]
  
logging.basicConfig(
    format="[%(asctime)s][%(levelname)s] %(message)s",
    level=LOG_LEVEL
)
  
logger = logging.getLogger()
  
logger.setLevel(LOG_LEVEL)
  
# GitHubアクセストークンを設定
access_token = config["access_token"]
  
# GitHubリポジトリとブランチ情報を設定
repository_name = config["repository_name"]
main_branch = config["main_branch"]
    
# BigQueryに接続する為のサービスアカウント等の設定  
credentials_file = config["credentials_file"]
credentials = service_account.Credentials.from_service_account_file(credentials_file)
bigquery_client = bigquery.Client(credentials=credentials)
  
# BigQueryのテーブル初期設定  
project = config["project"]
dataset_id = config["dataset_id"]
table_id = config["table_id"]
  
# GitHubに接続
g = Github(access_token)
  
  
# bqより編集したいidの必要なデータを取得する関数
def get_regist_data():
  
    query = f"select " \
            f"id, group_email, subnet_info, UPDATE_FLG, client_cidr, use_purpose " \
            f"from `{project}.{dataset_id}.{table_id}` " \
            f"where UPDATE_FLG is not null and use_purpose='normal' order by id asc"
  
    query_job = bigquery_client.query(query)
  
    # 必要なデータを収集
    data_group = []
    for row in query_job:
        _data = dict()
        for key, val in row.items():
            _data[key] = val
        data_group.append(_data)
  
    return data_group
  
  
# 申請内容に応じてブランチ及びproject_info.tfvarsファイルにデータを書き込みする関数
def create_branch(_res):
    # リポジトリを取得
    repo = g.get_repo(repository_name)
    # 新規ブランチ名
    new_branch_name = "stg_dev"
    # 更新ファイル(新規作成ファイル)
    updatefile = "update_contents.txt"
    # 削除ファイル
    deletefile = "test.txt"

    # 更新データ
    update_data = f'subnet_info : {_res["subnet_info"]}\n' \
                  f'group_email : {_res["group_email"]}\n' \
                  f'use_purpose : {_res["use_purpose"]}\n' \
                  f'client_cidr : {_res["client_cidr"]}'
  
    try:
        # 新しいブランチを作成
        repo.create_git_ref(ref=f'refs/heads/{new_branch_name}', sha=repo.get_branch(main_branch).commit.sha)
    except GithubException as e:
  
        if e.data["message"] == 'Reference already exists':
            pass
        else:
            # Error
            logger.error("exceptions : {} : {}".format(e, traceback.format_exc()))
            return 'NG'
  
    try:
        #　新規ファイル作成
        repo.create_file(
            path=updatefile,
            message="update_contents.txtファイルを作成",
            content=update_data,
            branch=new_branch_name
        )
  

        #  削除対象となるファイル全体をオブジェクトとして取得
        delete_contents = repo.get_contents(path=deletefile, ref=new_branch_name)
  
        #　ファイル削除
        repo.delete_file(
            path=deletefile,
            message=f"{deletefile}を削除",
            sha=delete_contents.sha,
            branch=new_branch_name
        )
    except GithubException as e:
  
        # 既にupdate_contents.txtファイルが存在する場合
        if e.data["message"] == "Invalid request.\n\n\"sha\" wasn't supplied.":
            update_file_class = repo.get_contents(updatefile, ref=new_branch_name)
            #　ファイル更新
            repo.update_file(
                path=updatefile,  # 転送元のブランチのファイルパス指定
                message=f"{updatefile}ファイルを更新",
                content=update_data,
                branch=new_branch_name,  # 送信先のブランチ指定
                sha=update_file_class.sha
            )
        else:
            # Error
            logger.error("exceptions : {} : {}".format(e, traceback.format_exc()))
            return 'NG'
  
    try:
        # プルリクエスト(stg_devブランチからmainブランチへ)
        pull_req = repo.create_pull(
            title="modify repository",
            body=f"Add updated {updatefile} file",
            head=new_branch_name,  # 新しいブランチを指定
            base=main_branch  # プルリクエストをマージする先のブランチを指定
        )
    except GithubException as e:
        # プルリクエストが既に存在する場合
        if 'A pull request already exists' in e.data["errors"][0]["message"]:
            pass
        else:
            logger.error("exceptions : {} : {}".format(e, traceback.format_exc()))
            return 'NG'
  
    return 'SUCCESS'
  
  
if __name__ == '__main__':
  
    res = get_regist_data()
  
    for row in res:
  
        if row["UPDATE_FLG"] == 'DLT':
            res = create_branch(row)
  
        if res == 'NG':
            logger.error(f'ID:{row["id"]}のブランチの更新に失敗しました!!')
            break