Backlogの課題にGitHubのコミットやプルリクをコメントとして入れたい！
やりたいことは、「Backlogの課題にGitHubのコミットやプルリクをコメントとして入れたい」です。
BacklogのGitを使っていればコミットコメントに課題キーがあればその課題のコメントにコミットが連携されます。
なのでGitHubを使っていても同じようにしたい！
[GithubとBacklogの連携] Backlogでissue管理して、Githubへのコミット内容をBacklogにも反映させる様に連携する方法 - Qiitaを見て簡単にできる！とおもったら・・・GitHubのServiceはWebhookに統合されて消えていました。
We have deprecated GitHub Services in favor of integrating with webhooks.
[https://developer.github.com/v3/guides/replacing-github-services/:title]
似たようなことをやっている人は世の中にいるのでいろいろ調べながら手作りすることにしました。
 |
|---|
| 流れはこんなイメージ |
GitHubとBacklogを連携するLambda関数を作る
1. BacklogでAPIキーを発行する
BacklogではAWSからやってくる処理を受け取るためのAPIキーを発行します。
 |
|---|
| 絵だとこの辺のことです |
- [個人設定]の画面を開く
- [API] > コメントを入力 > [登録]ボタンでAPIキーを発行する
2. AWSでLambdaとAPI Gatewayを作成する
AWSではGitHubからやってくる情報をAPI Gatewayで受け取ってLambda関数を呼び出して処理できるようにします。
 |
|---|
| 絵だとこの辺のことです |
IAM ポリシーを作成する
Lambda関数で使用する権限を作成します。
- [AWS マネジメントコンソール]から[IAM]の画面を開く
- サイドメニューの[ポリシー] > [ポリシーの作成]ボタンで作成画面を表示
- [JSON]タブを開いて以下のJSONを設定 > [ポリシーの確認]ボタンで内容を確認
- {アカウントID}に設定する値はAWS アカウント ID の確認方法 | AWSで確認する
- [名前]に任意の値を設定 > [ポリシーの作成]ボタンで作成する
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"Resource": "arn:aws:logs:対象のリージョン:{アカウントID}:*"
},
{
"Action": "secretsmanager:GetSecretValue",
"Effect": "Allow",
"Resource": "*"
}
]
}
IAM ロールを作成する
Lambda関数にアタッチするロールを作成してさっき作ったポリシーを設定します。
- [AWS マネジメントコンソール]から[IAM]の画面を開く
- サイドメニューの[ロール] > [ロールの作成]ボタンで作成画面を表示
- [AWSサービス] > [Lambda]を選択後に[次のステップ: アクセス権限]ボタンで次の画面を表示
- [ポリシーのフィルタ]で作成したポリシーを検索して選択後に[次のステップ: タグ]ボタンで次の画面を表示
- [タグの追加 (オプション)]は任意なので設定せずに[次のステップ: 確認]ボタンで次の画面を表示
- [ロール名]を入力して[ロールの作成]ボタンでロールを作成する
Lambda関数をとりあえず作る
まずは、実装を後回しにして関数だけ作ります。
- [AWS マネジメントコンソール]から[Lambda]の画面を開く
- [関数の作成]ボタンで作成画面を表示する
- [一から作成]を選択して以下を設定して[関数の作成]ボタンで関数を作成する
- 関数名 : 任意の名前(今回は
github_to_backlog) - ランタイム : Python3.7
- 実行ロール : 既存のロールを使用する
- 既存のロール : 作成したロールを選択
- 関数名 : 任意の名前(今回は
コードには初期コードがあるのでそのまま。実装は後でやります。
import json
def lambda_handler(event, context):
# TODO implement
return {
'statusCode': 200,
'body': json.dumps('Hello from Lambda!')
}
API Gatewayを作成する
- 参考
今回は、[HTTP API]と[REST API]のどちらを使おうか迷ったけれど、「使ったことがない」「GitHubのWebhookを受け取りたいだけ」なんといっても「低コストらしいぞ」という理由で[HTTP API]にしました。
- [AWS マネジメントコンソール]から[API Gateway]の画面を開く
- [APIを作成]ボタンで[APIの作成]画面を表示する
- [HTTP API]の[構築]ボタンで次の画面へ
- [統合を追加] > [Lambda] > [Lambda 関数]で作成したLambda関数を選択
- [API 名]に任意の名前を設定して[次へ]ボタンで[ルートを設定]画面へ
- 以下を設定して[次へ]ボタンで[ステージを定義]画面へ
- メソッド : POST
- リソースパス :
/{Lambda関数名} - 統合ターゲット : {Lambda関数名}
- [ステージを追加]ボタンで以下を追加して[次へ]ボタンで[確認して作成]画面へ
- ステージ名 :
$default - 自動デプロイ : ON
- ステージ名 :
- [作成]ボタンで作成する
- 「{Lambda関数名}のステージ」一覧の[URLを呼び出す]列に「呼び出しURL」が表示される
- curlコマンドを使ってAPI Gatewayを呼び出してAPI GatewayがLambda関数を呼び出せることを確認する
| curlコマンドのオプション | 意味 |
|---|---|
| -X | HTTPメソッドを指定する |
| -H | HTTPヘッダを指定する |
# Lambda関数の初期コードに書いてある「Hello from Lambda!」が返却される
$ curl -X POST -H 'Content-Type:application/json' {呼び出しURL}/{リソースパス}
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
}
100 20 100 20 0 0 50 0 --:--:-- --:--:-- --:--:-- 50"Hello from Lambda!"
# 失敗例) 「リソースパス」をくっつけ忘れると「Not Found」になるので注意
$ curl -X POST -H 'Content-Type:application/json' {呼び出しURL}
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
100 23 100 23 0 0 287 0 --:--:-- --:--:-- --:--:-- 291{"message":"Not Found"}
3. GithubでWebhook設定する
GitHubでのプッシュやプルリクの情報がAPI Gatewayに送られるようにWebhookを設定します。
 |
|---|
| 絵だとこの辺のことです |
- ブラウザでGitHubのリポジトリを表示する
- [Settings] > [Webhooks] > [Add webhook]ボタン
- 以下を設定して[Add webhook]ボタン
- Payload URL : API Gatewayの「呼び出しURL/リソースパス」
- Content type : application/json
- Secret : 推測されにくい任意の文字列
- SSL verification : Enable SSL verification
- Which events would you like... : 「Let me select individual events.」にして以下を選択
- Pull requests
- Pushes
- 表示されたWebhookの横につくマークが緑チェックになるのを確認する
- URLが間違っていたりすると赤バツが付くので、その場合は内容を確認する
4. AWSでSecrets Managerに情報を登録する
BacklogのAPIキーやGitHubのWebhookに設定したSecretは大切な情報なのでSecrets Managerに登録して、Lambda関数から取得して使うようにします。
 |
|---|
| 絵だとこの辺のことです |
- [新しいシークレットを保存する]ボタンから作成画面を表示して以下を設定して[次]ボタン
- シークレットの種類: その他のシークレット
- シークレットのペア: 下記表参照
- 暗号化キー: DefaultEncryptionKey
- 以下を設定して[次]ボタン
- シークレットの名前: 下記表参照
- 説明とタグ: 任意
- [自動ローテーションを無効にする]を設定して[次]ボタン
- [保存]ボタンでシークレットを作成する
| シークレットの名前 | シークレットキー | シークレットの値 | 使うところ |
|---|---|---|---|
| github/to/backlog | GITHUB_SECRET | GitHubのWebhookに 設定したSecret |
GitHubから来た情報 をHMAC認証する時に使う |
| backlog/{GitHubのusername} | APIKEY | BacklogのAPIキー | Backlogに各ユーザで コメント追加するのに使う |
| 同上 | Backlogに登録され ているメールアドレス (以下手順で確認) 1.Backlogの[個人設定]の画面を開く 2. [ユーザー情報] > [メールアドレス] |
プルリクエストの通知を つけるためのid取得に使う |
5. Lambda関数を実装する
環境変数を設定する
Backlogの課題にコメント追加するにはBacklog APIを使用します。
そのためにBacklog APIの情報としてLambdaの環境変数に以下を設定して使用します。
| 環境変数のキー | 値 | 説明 |
|---|---|---|
| BACKLOG_ENDPOINT | BacklogAPIのエンドポイント |
https://BacklogのURLと同じ値/api/v2/参考:認証と認可 | Backlog Developer API | Nulab |
| PROJECT_KEY | Backlogのプロジェクトキー | 参考 : プロジェクトの追加 – Backlog ヘルプセンター |
Secrets Managerからシークレットの値を取得する
登録したBacklogのAPIキーやGitHubのWebhookに設定したSecretを取得できるようにします。
基本的な実装はシークレットを登録した際にSecrets Managerの画面に表示されるサンプルコードを使用しました。
def get_secrets_manager_dict(secret_name: str) -> dict:
"""Secrets Managerからシークレットのセットを辞書型で取得する"""
secrets_dict = {}
if not secret_name:
print('シークレットの名前未設定')
else:
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name='対象のリージョン'
)
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
except ClientError as e:
print('シークレット取得失敗：シークレットの名前={}'.format(secret_name))
print(e.response['Error'])
else:
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
else:
secret = base64.b64decode(get_secret_value_response['SecretBinary'])
secrets_dict = ast.literal_eval(secret)
return secrets_dict
GitHubのWebhookから送られてくる内容を取得する処理を作る
GitHubのWebhookから送られてくる内容は以下サイトに説明があります。
まずは、GitHubから送られてきた情報が本当に設定したWebhookからなのを確認するためにHMAC認証します。
GitHubのWebhookに設定したSecretとGitHubから送られてきた情報で認証を行います。
参考 : GitHubのWebhookでプルリクエストをマージした際にツイートできるようしてみた - Qiita
def is_correct_signature(signature: str, body: dict) -> bool:
"""GitHubから送られてきた情報をHMAC認証する."""
if signature and body:
# GitHubのWebhookに設定したSecretをSecrets Managerから取得する
secret = get_secrets_manager_key_value('github/to/backlog', 'GITHUB_SECRET')
if secret:
secret_bytes = bytes(secret, 'utf-8')
body_bytes = bytes(body, 'utf-8')
# Secretから16進数ダイジェストを作成する
signedBody = "sha1=" + hmac.new(secret_bytes, body_bytes, hashlib.sha1).hexdigest()
return signature == signedBody
else:
return False
プッシュとプルリクではGitHubからくる情報が異なるというのとBacklogの課題に追加するコメントをちょっと変えるために処理を切り分けます。
| 操作 | action | pull_request | commits |
|---|---|---|---|
| Pull requests | o | o | x |
| Pushes | x | x | o |
def lambda_handler(event, context):
# GitHubから送られてきた情報をHMAC認証する
if is_correct_signature(event['headers']['x-hub-signature'], event['body']):
body = json.loads(event['body'])
# プッシュとプルリクを識別して処理を切り分ける
if 'pull_request' in body and 'action' in body:
# Pull requestsの場合
add_pull_request_comment(body['action'], body['pull_request'], body['sender']['login'])
elif 'commits' in body:
# Pushesの場合
add_push_comment(body['commits'])
else:
print('処理対象外のリクエストなので処理しない' + json.dumps(body))
else:
print('認証できないGitHubのsignatureが送られてきた')
コメントから課題キーを検索する
コメント追加する課題を決めるためにコミットコメントやプルリクのコメントから課題キーを検索します。
複数の課題キーがあればそれぞれの課題にコメントが追加できるようにリストに課題キーを入れて返却します。
def get_issue_key(message: str) -> list:
"""コメントから課題キーを検索する."""
issue_key = []
# 「[プロジェクトキー] + [-] + [数字の繰り返し]を全て抜出
key_format = '{}-[\d]+'.format(os.environ.get('PROJECT_KEY'))
match_list = re.findall(key_format, message)
if match_list:
# 抜き出した課題キーのリストから重複を削除する
issue_key = list(set(match_list))
else:
print('コメントにBacklogの課題キーが設定されていない')
return issue_key
プルリクエストはレビューアーに通知をつけたいので通知リストを作る
通知リストは次の流れで作成します。
- プルリクエストに設定されたレビューアーのGitHubユーザー名でSecrets ManagerからBacklogのメールアドレスを取得する
- APIで取得したBacklogのユーザー一覧からメールアドレスでBacklogのユーザーIDを探す
- 通知リストへ追加する
def create_notified_list(requested_reviewers: list, backlog_users: list) -> list:
"""レビューアーにBacklogの通知をつけるため、にコメント登録の通知を受け取るユーザーIDリストを作成する"""
notified_list = []
if requested_reviewers and backlog_users:
for reviewer in requested_reviewers:
# Secrets ManagerからBacklogのメールアドレスを取得する
backlog_mail = get_secrets_manager_key_value('backlog/' + reviewer['login'], 'MAIL')
if backlog_mail:
for user in backlog_users:
# 取得したメールアドレスと同じメールアドレスのユーザーをBacklogユーザー一覧から探す
if backlog_mail == user['mailAddress']:
# ユーザーのIDをリストへ追加する
notified_list.append(user['id'])
return notified_list
Backlogにコメントを追加する
def add_backlog_comment(api_key: str, issue_keys: list, comment: str, notified_list: list):
"""Backlogの課題にコメントを追加する."""
if not api_key or not issue_keys:
print('BacklogのAPIキーまたは課題キー未設定')
else:
params = {'apiKey': api_key}
payload = {'content': comment}
# コメント登録の通知を受け取るユーザーIDがある場合は設定する
if notified_list:
payload['notifiedUserId[]'] = notified_list
# コメントにある課題すべてにコメントを追加する
for issue_key in issue_keys:
header = {'Content-Type': 'application/x-www-form-urlencoded'}
api_path = urllib.parse.urljoin(os.environ.get('BACKLOG_ENDPOINT'), '/'.join(['issues', issue_key, 'comments']))
result = requests.post(api_path, headers=header, params=params, data=payload)
できた!!
今回はシンプルにコメント追加をしました。
今後、BacklogのGitみたいに課題のステータス変更をしたり、レビューコメントも追加したりしたら楽しそうです！
コードの全体
ここ以降は、コード全部を張っているだけなので興味のある人だけ見てください。
import json, os
import hmac, hashlib
import requests, urllib
import boto3
import base64
import ast, re
from botocore.exceptions import ClientError
def get_secrets_manager_dict(secret_name: str) -> dict:
"""Secrets Managerからシークレットのセットを辞書型で取得する"""
secrets_dict = {}
if not secret_name:
print('シークレットの名前未設定')
else:
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name='対象のリージョン'
)
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
except ClientError as e:
print('シークレット取得失敗：シークレットの名前={}'.format(secret_name))
print(e.response['Error'])
else:
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
else:
secret = base64.b64decode(get_secret_value_response['SecretBinary'])
secrets_dict = ast.literal_eval(secret)
return secrets_dict
def get_secrets_manager_key_value(secret_name: str, secret_key: str) -> str:
"""AWS Secrets Managerからシークレットキーの値を取得する."""
value = ''
secrets_dict = get_secrets_manager_dict(secret_name)
if secrets_dict:
if secret_key in secrets_dict:
# secrets_dictが設定されていてsecret_keyがキーとして存在する場合
value = secrets_dict[secret_key]
else:
print('シークレットキーの値取得失敗：シークレットの名前={}、シークレットキー={}'.format(secret_name, secret_key))
return value
def is_correct_signature(signature: str, body: dict) -> bool:
"""GitHubから送られてきた情報をHMAC認証する."""
if signature and body:
# GitHubのWebhookに設定したSecretをSecrets Managerから取得する
secret = get_secrets_manager_key_value('github/to/backlog', 'GITHUB_SECRET')
if secret:
secret_bytes = bytes(secret, 'utf-8')
body_bytes = bytes(body, 'utf-8')
# Secretから16進数ダイジェストを作成する
signedBody = "sha1=" + hmac.new(secret_bytes, body_bytes, hashlib.sha1).hexdigest()
return signature == signedBody
else:
return False
def get_backlog_api_key(github_username: str) -> str:
"""GitHubのユーザ名から該当ユーザのBacklogのAPIキーを取得する."""
api_key = ''
if github_username:
api_key = get_secrets_manager_key_value('backlog/' + github_username, 'APIKEY')
else:
print('GitHubユーザー名未設定')
return api_key
def add_backlog_comment(api_key: str, issue_keys: list, comment: str, notified_list: list):
"""Backlogの課題にコメントを追加する."""
if not api_key or not issue_keys:
print('BacklogのAPIキーまたは課題キー未設定')
else:
params = {'apiKey': api_key}
payload = {'content': comment}
# コメント登録の通知を受け取るユーザーIDがある場合は設定する
if notified_list:
payload['notifiedUserId[]'] = notified_list
# コメントにある課題すべてにコメントを追加する
for issue_key in issue_keys:
header = {'Content-Type': 'application/x-www-form-urlencoded'}
api_path = urllib.parse.urljoin(os.environ.get('BACKLOG_ENDPOINT'), '/'.join(['issues', issue_key, 'comments']))
result = requests.post(api_path, headers=header, params=params, data=payload)
def get_backlog_users(api_key: str) -> list:
"""Backlogユーザー一覧の取得"""
users = []
if not api_key:
print('BacklogのAPIキー未設定')
else:
# ユーザーの取得対象はプロジェクト内に設定する
api_path = urllib.parse.urljoin(os.environ.get('BACKLOG_ENDPOINT'), '/'.join(['projects', os.environ.get('PROJECT_KEY'), 'users']))
api_result = requests.get(api_path, params={'apiKey': api_key}).json()
if type(api_result) == list and api_result:
# Backlogのユーザー一覧を取得する
users = api_result
else:
print('Backlogプロジェクトのユーザー一覧取得失敗：{}'.format(json.dumps(api_result)))
return users
def create_notified_list(requested_reviewers: list, backlog_users: list) -> list:
"""レビューアーにBacklogの通知をつけるため、にコメント登録の通知を受け取るユーザーIDリストを作成する"""
notified_list = []
if requested_reviewers and backlog_users:
for reviewer in requested_reviewers:
# Secrets ManagerからBacklogのメールアドレスを取得する
backlog_mail = get_secrets_manager_key_value('backlog/' + reviewer['login'], 'MAIL')
if backlog_mail:
for user in backlog_users:
# 取得したメールアドレスと同じメールアドレスのユーザーをBacklogユーザー一覧から探す
if backlog_mail == user['mailAddress']:
# ユーザーのIDをリストへ追加する
notified_list.append(user['id'])
return notified_list
def get_issue_key(message: str) -> list:
"""コメントから課題キーを検索する."""
issue_key = []
# 「[プロジェクトキー] + [-] + [数字の繰り返し]を全て抜出
key_format = '{}-[\d]+'.format(os.environ.get('PROJECT_KEY'))
match_list = re.findall(key_format, message)
if match_list:
# 抜き出した課題キーのリストから重複を削除する
issue_key = list(set(match_list))
else:
print('コメントにBacklogの課題キーが設定されていない')
return issue_key
def add_push_comment(commits: list):
print('プッシュの情報をBacklogの課題へコメント追加する')
print(json.dumps(commits))
# プッシュに含まれるコミットを1つずつ処理する
for commit in commits:
issue_key = []
if 'message' in commit:
issue_keys = get_issue_key(commit['message'])
if issue_keys:
backlog_api_key = get_backlog_api_key(commit['committer']['username'])
if backlog_api_key:
comment_format = '- URL：{}\n{}'
comment = comment_format.format(commit['url'], commit['message'])
add_backlog_comment(backlog_api_key, issue_keys, comment, [])
def add_pull_request_comment(action: str, pull_request: dict, username: str):
print('プルリクエストの情報をBacklogの課題へコメント追加する\nアクション：{}\nマージ：{}'.format(action, str(pull_request['merged'])))
print(json.dumps(pull_request))
# プルリクのタイトルとコメントから課題キーを取得する
issue_keys = get_issue_key(pull_request['title'] + pull_request['body'])
if issue_keys:
# BacklogのAPIキーを取得する
backlog_api_key = get_backlog_api_key(username)
if backlog_api_key:
notified_list = []
if 'requested_reviewers' in pull_request and pull_request['requested_reviewers']:
# Backlogのユーザー一覧を取得する
backlog_users = get_backlog_users(backlog_api_key)
if backlog_users:
# レビューアーが設定されている場合は追加するコメント用の通知リストを作成する
notified_list = create_notified_list(pull_request['requested_reviewers'], backlog_users)
# プルリクのアクションによってコメントを作成する
comment = ''
if action == 'opened' or action == 'reopened':
comment = 'プルリクエストが作成されました'
elif action == 'closed':
if pull_request['merged']:
comment = 'プルリクエストがマージされました'
else:
comment = 'プルリクエストが却下されました'
else:
comment = 'プルリクエストが変更されました'
comment += '\n\n- URL：{}'.format(pull_request['html_url'])
add_backlog_comment(backlog_api_key, issue_keys, comment, notified_list)
def lambda_handler(event, context):
# GitHubから送られてきた情報をHMAC認証する
if is_correct_signature(event['headers']['x-hub-signature'], event['body']):
body = json.loads(event['body'])
# プッシュとプルリクを識別して処理を切り分ける
if 'pull_request' in body and 'action' in body:
# Pull requestsの場合
add_pull_request_comment(body['action'], body['pull_request'], body['sender']['login'])
elif 'commits' in body:
# Pushesの場合
add_push_comment(body['commits'])
else:
print('処理対象外のリクエストなので処理しない' + json.dumps(body))
else:
print('認証できないGitHubのsignatureが送られてきた')