G-gen の佐伯です。 前編 では PyGithub の管理タスク自動化の事前準備・導入について紹介しました。後編では新規ブランチの作成、ブランチ内ファイルの更新・削除、プルリクエストの自動化についてご紹介したいと思います。 はじめに 前編記事 当記事でやること 事前準備 データを準備 Github リポジトリを作成・main ブランチにファイルを準備 新規ブランチ作成 ファイルの新規作成 ファイルの更新 ファイルの削除 プルリクエスト プログラムの実行 ソースコード 実行結果 はじめに 前編記事 当記事では、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 ) ファイルの更新 update_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 実行結果 想定通り stg_dev ブランチが作成され、 update_contents.txt ファイルにデータが書き込まれ、 main ブランチにプルリクが出されています。 ①プルリク ②ファイルの更新 stg_dev ブランチ内の test.txt ファイルは削除され、 update_contents.txt ファイルにデータが書き込まれています。 ③stg_devブランチの作成 佐伯 修 (記事一覧) クラウドソリューション部 前職では不動産業でバックエンドを経験し、2022年12月G-genにジョイン。 入社後、Google Cloudを触り始め、日々スキル向上を図る。 SEの傍ら、農業にも従事。水耕を主にとする。