こんにちは。
マイクロアドのタカギです。
slackのfiles.upload API が廃止になるとのことで、対応が必要になるため、サンプルコードを作ってみました。
公式SDKを使用する場合はfiles.upload V2を使用すれば問題ないようですが、今回もrequestsモジュールを使用してSlack通知してみようと思います。
やること
前回と同じことをやります。
- slackにメッセージを投稿する
- 投稿したメッセージのスレッドにファイルを投稿する
※前回
準備
まずslackのトークンを環境変数にセットします。
これでうっかりトークンがインターネットに公開されてしまうリスクを減らします。
export SLACK_BOT_TOKEN="xxxxx"
サンプルコード
省略していますが、mypyとかも使っています。
.python-version
3.11.7
pyproject.toml
[tool.poetry]
package-mode = false
[tool.poetry.dependencies]
python = "3.11.7"
requests = "2.31.0"
app.py
import json
import logging
import os
import uuid
import requests # type: ignore
LOGGER = logging.getLogger('app')
LOGGER.setLevel('INFO')
logging.basicConfig(level='INFO', format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
SLACK_BOT_TOKEN = os.environ.get('SLACK_BOT_TOKEN', '')
channel_id = 'xxx'
def main() -> None:
"""
slack通知のサンプルコード
"""
slack_msg = 'slackメッセージテスト'
file_title = 'ファイルタイトルテスト'
file_content = '''
ファイル内容
改行も含みます
日本語も含みます
'''
# slackにメッセージを投稿する
response_json = send_message(slack_msg, channel_id, SLACK_BOT_TOKEN)
# メッセージ通知したスレッドにファイル形式で通知
thread_ts = response_json['ts']
send_file_content_on_message_thread(file_title, file_content, channel_id, thread_ts, SLACK_BOT_TOKEN)
def send_message(msg: str, channel_id: str, token: str) -> dict:
"""
slackにメッセージを投稿する
Args:
msg: 通知メッセージ
channel_id: slackのチャンネルID
token: slackトークン
Returns:
APIのレスポンス情報
"""
LOGGER.info('send message to slack.')
url = 'https://slack.com/api/chat.postMessage'
header = {'Authorization': f'Bearer {token}'}
data = {'channel': channel_id, 'text': msg}
response = requests.post(url, headers=header, data=data)
response.raise_for_status()
if not response.ok:
raise Exception('failed to post messeage to slack.')
response_json = response.json()
LOGGER.info(response_json)
return response_json
def send_file_content_on_message_thread(
file_title: str, file_content: str, channel_id: str, thread_ts: str, token: str
) -> None:
"""
slackの指定スレッドにファイルを投稿する
Args:
file_title: ファイルタイトル
file_content: ファイル内容
channel_id: slackのチャンネルID
thread_ts: スレッドのタイムスタンプ
token: slackトークン
"""
# ファイル名はランダムな文字列
file_name = f'{str(uuid.uuid4())}.txt'
# 投稿するファイル内容のサイズを取得する
length = len(file_content.encode(encoding='utf-8'))
# ファイルアップロードURLとファイルIDを取得する
upload_url, file_id = _get_upload_url_and_file_id(file_name, length, token)
# ファイルをアップロードする
_upload_file_content(upload_url, file_content)
# ファイルアップロードを完了する
_complete_upload(file_id, file_title, channel_id, thread_ts, token)
def _get_upload_url_and_file_id(filename: str, length: int, token: str) -> tuple[str, str]:
"""
ファイルアップロードURLとファイルIDを取得する
Args:
filename: アップロードファイル名
length: アップロードファイルサイズ
token: slackトークン
Returns:
APIのレスポンス情報
ファイルアップロードURLとファイルIDのタプル
"""
LOGGER.info('get upload url from slack.')
url = 'https://slack.com/api/files.getUploadURLExternal'
header = {'Authorization': f'Bearer {token}'}
params = {'filename': filename, 'length': length}
response = requests.get(url, headers=header, params=params)
response.raise_for_status()
if not response.ok:
raise Exception('failed to get upload url from slack')
response_json = response.json()
LOGGER.info(response_json)
return response_json['upload_url'], response_json['file_id']
def _upload_file_content(upload_url: str, file_content: str) -> None:
"""
ファイルをアップロードする
Args:
upload_url: ファイルアップロードURL
file_content: ファイル内容
"""
LOGGER.info('upload file to slack.')
data = file_content.encode(encoding='utf-8')
response = requests.post(upload_url, data=data)
response.raise_for_status()
if not response.ok:
raise Exception('failed to upload file to slack.')
def _complete_upload(file_id: str, title: str, channel_id: str, thread_ts: str, token: str) -> None:
"""
ファイルアップロードを完了する
Args:
file_id: ファイルID
title: ファイルのタイトル
channel_id: slackのチャンネルID
thread_ts: スレッドのタイムスタンプ
token: slackトークン
"""
LOGGER.info("complete upload.")
url = 'https://slack.com/api/files.completeUploadExternal'
header = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json; charset=utf-8'}
data = {'files': [{'id': file_id, 'title': title}], 'channel_id': channel_id, 'thread_ts': thread_ts}
response = requests.post(url, headers=header, data=json.dumps(data))
response.raise_for_status()
if not response.ok:
raise Exception('failed to complete upload.')
if __name__ == '__main__':
main()
環境構築
poetryを使うとスムーズですね。
poetry install
poetry shell
実行
いざ！
python app.py
実行結果
メッセージが通知されて、
スレッドにも通知されました。
まとめ
requestsモジュールを使用して、files.upload API を使用せずに、slackにメッセージ投稿とファイル投稿ができました。
お疲れ様でした。