概要
Google Driveのフォルダ内にあるスプレッドシート情報をSlackに通知するCloud Functionsを試験的に作成してみたのでご紹介します。本記事は実装編です。
セットアップ編はこちら
構成
- Google Cloud Functions
- Google Cloud Run
- Google Drive API
- Google Sheets API
- Slack Webhook
全体の流れ
以下の流れで記事を進めています。本記事は3,4の紹介です。
- Google Cloudの設定:
- サービスアカウントの作成と必要なロールの付与
- Workload Identity連携の設定
- Slack Webhookの設定
-
Cloud Functionsの実装
- Google Drive APIとGoogle Sheets APIを使用して、フォルダ内のファイルリストをファイル最終更新者にメンションするFunctionを作成
- Cloud Runでの実行と結果
検証用Driveの作成
Cloud Functions実装の前に、Driveへ検証用フォルダを作成しセットアップ編で作成したサービスアカウントに共有します。
フォルダIDは実装で使用します。
https://drive.google.com/drive/folders/ここからフォルダID
Cloud Functionsの作成
「関数を作成」から設定をしてFunctionを作成していきます。
- 関数名と認証設定
- 「認証が必要」にしました
- ランタイムとランタイムサービスアカウントに以下を設定：
- ランタイム：180
- ランタイムサービスアカウント：セットアップ編で作成したサービスアカウント
- Build サービス アカウント
- 「カスタム サービス アカウント」を選択しセットアップ編で作成したサービスアカウント設定
- 接続：デフォルト設定
- セキュリティとイメージのリポジトリ：デフォルト設定
Cloud Functionsの実装内容
今回は検証的に以下の流れで実装しています
- Google Driveフォルダ内のファイルリストを取得(今回はスプレッドシートのみ)
- スプレッドシート最終更新者のメールアドレスを取得
- Slackでスプレッドシート最終更新者にメンションしフォルダ内のファイルリストを通知
from google.auth import default
from googleapiclient.discovery import build
import requests
import json
# 認証情報を取得
def get_credentials():
credentials, project = default()
return credentials
# Google Drive API設定
def create_drive_service():
credentials = get_credentials()
return build('drive', 'v3', credentials=credentials)
# Google Sheets API設定
def create_sheets_service():
credentials = get_credentials()
return build('sheets', 'v4', credentials=credentials)
# 指定したフォルダ内のスプレッドシートIDを取得
def get_spreadsheet_id(folder_id):
drive_service = create_drive_service()
query = f"'{folder_id}' in parents and mimeType='application/vnd.google-apps.spreadsheet'"
results = drive_service.files().list(q=query, pageSize=1).execute()
items = results.get('files', [])
if not items:
raise ValueError('No spreadsheet found in the specified folder.')
else:
return items[0]['id']
# スプレッドシートの最終更新者のメールアドレスを取得
def get_last_updated_user_email(spreadsheet_id):
drive_service = create_drive_service()
revisions = drive_service.revisions().list(fileId=spreadsheet_id, pageSize=1, fields="revisions(lastModifyingUser)").execute()
last_revision = revisions.get('revisions', [])[0]
last_updated_user = last_revision['lastModifyingUser']
return last_updated_user['emailAddress']
# 指定したフォルダ内のファイルリストを取得
def list_files():
drive_service = create_drive_service()
folder_id = 'フォルダIDを指定'
query = f"'{folder_id}' in parents"
results = drive_service.files().list(q=query, pageSize=10).execute()
items = results.get('files', [])
if not items:
return 'No files found.'
else:
files_list = 'Files:\n'
for item in items:
files_list += f"{item['name']} ({item['id']})\n"
return files_list
# Slackに通知を送信する関数
def send_slack_notification(message, email):
webhook_url = 'Webhook URLを指定'
slack_data = {
'text': message,
'blocks': [
{
'type': 'section',
'text': {
'type': 'mrkdwn',
'text': f'<@{email}>: {message}' # メンション形式
}
}
]
}
response = requests.post(
webhook_url, data=json.dumps(slack_data),
headers={'Content-Type': 'application/json'}
)
if response.status_code != 200:
raise ValueError(f'Request to Slack returned an error {response.status_code}, the response is:\n{response.text}')
# Cloud Functionのエントリーポイントとなる関数
def cloud_function(request):
response = list_files()
print(response)
# フォルダIDを指定してスプレッドシートIDを取得
folder_id = 'フォルダIDを指定'
spreadsheet_id = get_spreadsheet_id(folder_id)
# スプレッドシートIDを指定して最終更新者のメールアドレスを取得
email = get_last_updated_user_email(spreadsheet_id)
send_slack_notification(response, email)
return response
Cloud Functionsをデプロイ後、Cloud Runで動作確認
作成したFunctionをコンソールでデプロイ後、CloudShellからcurlでFunctionを実行。
想定通りドライブ内のスプレッドシート名とスプレッドシートIDがSlack通知されました。
以上です。
今回使用したAPIのドキュメントは以下です。どなたかのお役に立てれば幸いです。
Google Drive API
Google Sheets API