KATOエンジニヤリング開発日誌

「アウトプット無きエンジニアにインプットもチャンスも無い」の精神で書いています

ConoHa APIをつかって「お知らせ」を定期的に確認する - 実装編 -

前回は使用するConoHa APIの詳細について調査しました。

www.kato-eng.info

今回は実際にプログラムの内容について解説していきます。

ファイル構成

スクリプト自体はいつも通りPythonで実装しています。

ディレクトリ構成は下記の通りです。

f:id:masayuki_kato:20190706184803p:plain
ファイル構成

  • config (ディレクトリ)
    • account.ini
      • Slack通知を行うための設定が記載されている
    • conoha.ini
      • ConoHa APIのエンドポイントや各種設定が記載されている
  • conoha (ディレクトリ)
    • conoha_config.py
      • ConoHa API系のスクリプトログ設定が記載されている
    • functions (ディレクトリ)
      • api_common.py
        • ConoHa API系の共通関数が格納されているファイル
    • get_notice.py
      • メインスクリプト

プログラム内容

以降のスクリプトや設定ファイルでは今回の処理に関係無い部分はいくつか省略しています。

※ Github上には全て公開しています。

github.com

config ディレクトリ

account.ini
# Slack Information
[slack]
slack_log_url = Slack WebhookのURL
conoha.ini
# common
[common]    # ConoHa APIで使用するユーザー情報
TENANT_ID = conohaのテナント情報
USER_NAME = conohaのユーザー名
PASSWARD = conohaのログインパスワード

# Header
[header]    # APIで使用する「Header」項目
ACCEPT_JSON = application/json

# Base Endpoint
[base_endpoint]    # APIのエンドポイントでConoHaのコントロールパネルから確認できる
ENDPOINT_IDENTITY_SERVICE = https://identity.tyo1.conoha.io/v2.0
ENDPOINT_ACCOUNT_SERVICE = https://account.tyo1.conoha.io/v1

# Endpoint Path    # 上記「base_endpoint」の値の後に追加するエンドポイント
[endpoint]
TOKENS = /tokens
NOTIFICATIONS = /notifications

conoha ディレクトリ

conoha_config.py

conohaディレクトリに格納されている各スクリプトのロギング機能の設定ファイル。

プログラムのロジックには影響が無いので割愛します。

functions/api_common.py
#coding:utf-8

import requests
import json

# Constants
SLASH = "/"

# トークン取得API
def get_token(BASE_ENDPOINT, SUB_PATH, TENANT_ID, USER_NAME, PASSWARD, ACCEPT_JSON):
    """
    BASE_ENDPOINT : APIのエンドポイント
    SUB_PATH : BASE_ENDPOINTに追加するエンドポイント
    TENANT_ID : ConoHa コントロールパネルから確認できるテナントID
    USER_NAME : ConoHa コントロールパネルから確認できるユーザーID
    PASSWARD : ConoHa コントロールパネルにログインするためのパスワード
    ACCEPT_JSON : APIのヘッダ「Accept」にセットする値
    """

    url = BASE_ENDPOINT + SUB_PATH
    # ボディ部の作成
    body = {
        "auth":{
            "passwordCredentials":{
                "username": USER_NAME,
                "password": PASSWARD
            },
        "tenantId": TENANT_ID
        }
    }

    # POSTメソッドでAPIを実行
    respons = requests.post(
        url,
        json.dumps(body),
        headers={'Accept': ACCEPT_JSON}
    )

    # HTTPステータスコードが「200:OK」
    if respons.status_code == requests.codes.ok:
        respons_json = json.loads(respons.text)
        return respons_json['access']['token']['id']
    # HTTPステータスコードが200以外の場合はエラーとする
    else:
        return respons.raise_for_status()

# 告知一覧取得API
def get_notification_code_list(BASE_ENDPOINT, SUB_PATH, TENANT_ID, ACCEPT_JSON, token):
    """
    BASE_ENDPOINT : APIのエンドポイント
    SUB_PATH :  BASE_ENDPOINTに追加するエンドポイント
    TENANT_ID : ConoHa コントロールパネルから確認できるテナントID
    ACCEPT_JSON : APIのヘッダ「Accept」にセットする値
    token : トークン取得APIで取得したAPIアクセストークン
    """
    url = BASE_ENDPOINT + SLASH + TENANT_ID + SUB_PATH
    # GETメソッドでAPI実行
    respons = requests.get(
        url,
        params={'limit': 20},
        headers={'Accept': ACCEPT_JSON, 'X-Auth-Token': token}
    )
        # HTTPステータスコードが「200:OK」
    if respons.status_code == requests.codes.ok:
        respons_json = json.loads(respons.text)
        return respons_json['notifications']
    # HTTPステータスコードが200以外の場合はエラーとする
    else:
        return respons.raise_for_status()

# 告知既読・未読変更API
def set_read_status(BASE_ENDPOINT, SUB_PATH, TENANT_ID, ACCEPT_JSON, token, notice_code, read_status):
    """
    BASE_ENDPOINT : APIのエンドポイント
    SUB_PATH :  BASE_ENDPOINTに追加するエンドポイント
    TENANT_ID : ConoHa コントロールパネルから確認できるテナントID
    ACCEPT_JSON : APIのヘッダ「Accept」にセットする値
    token : トークン取得APIで取得したAPIアクセストークン
    notice_code : 告知一覧取得APIで取得した「告知コード」
    read_status : 告知の「既読・未読 状態ステータス」
    """
    url = BASE_ENDPOINT + SLASH + TENANT_ID + SUB_PATH + SLASH + str(notice_code)
    # ボディ部の生成
    body = {
        "notification":{
            "read_status": read_status
        }
    }
    # PUTメソッドでAPI実行
    response = requests.put(
        url,
        json.dumps(body),
        headers={'Accept': ACCEPT_JSON, 'X-Auth-Token': token}
    )
    if response.status_code == requests.codes.ok:
        response_json = json.loads(response.text)
        return response_json['notification']['read_status']
    else:
        return response.raise_for_status()
get_notice.py
#coding:utf-8

import conoha_config
import logging
import slackweb
import configparser
from functions import api_common
import requests
import sys

# Slack通知の設定ファイル読み込み
BASE_DIR = '/usr/local/script/'
account = BASE_DIR + 'config/account.ini'
account_config = configparser.ConfigParser()
account_config.read(account, 'UTF-8')

# ConoHa APIで使用する設定ファイル読み込み
conoha = BASE_DIR + 'config/conoha.ini'
conoha_config = configparser.ConfigParser()
conoha_config.read(conoha, 'UTF-8')

# Constants
TENANT_ID = conoha_config.get('common', 'TENANT_ID')
USER_NAME = conoha_config.get('common', 'USER_NAME')
PASSWARD = conoha_config.get('common', 'PASSWARD')
ACCEPT_JSON = conoha_config.get('header', 'ACCEPT_JSON')
IDENTITY_ENDPOINT = conoha_config.get('base_endpoint', 'ENDPOINT_IDENTITY_SERVICE')
ACCOUNT_ENDPOINT = conoha_config.get('base_endpoint', 'ENDPOINT_ACCOUNT_SERVICE')
TOKEN_PATH = conoha_config.get('endpoint', 'TOKENS')
NOTIFICATIONS = conoha_config.get('endpoint', 'NOTIFICATIONS')


if __name__ == "__main__":
    # モジュール名でロガーを生成する(メインモジュールは 名前が '__main__' になる)
    log = logging.getLogger(__name__)
    # Slack Incoming webhook設定
    slack_log_url = account_config.get('slack', 'slack_log_url')
    slack = slackweb.Slack(url=slack_log_url)

    log.info('ConoHa告知一覧取得 : 開始')

    # APIトークン取得
    try:
        token = api_common.get_token(IDENTITY_ENDPOINT, TOKEN_PATH, TENANT_ID, USER_NAME, PASSWARD, ACCEPT_JSON)
    except requests.RequestException as e:
        log.error("APIトークン取得 失敗 : " + str(e))
        slack.notify(text="【ERROR】ConoHa 告知情報取得処理失敗しました。ログを確認してください。")
        sys.exit(1)

    # 告知一覧取得処理
    try:
        notice_list = api_common.get_notification_code_list(ACCOUNT_ENDPOINT, NOTIFICATIONS, TENANT_ID, ACCEPT_JSON, token)
    except requests.RequestException as e:
        log.error("告知一覧取得処理 失敗 : " + str(e))
        slack.notify(text="【ERROR】ConoHa 告知情報取得処理失敗しました。ログを確認してください。")
        sys.exit(1)

    title_list = list()
    # 各告知をslack通知
    for notice in notice_list:
        if notice['read_status'] == 'Read':
            continue
        title_list.append(notice['title'])
        # 告知ステータスを既読に変更
        try:
            notice_code = notice['notification_code']
            read_status = 'Read'
            chenge_status = api_common.set_read_status(
                ACCOUNT_ENDPOINT,
                NOTIFICATIONS,
                TENANT_ID,
                ACCEPT_JSON,
                token,
                notice_code,
                read_status
            )
            if chenge_status != read_status:
                log.warn("既読ステータス変更失敗 : " + notice['title'])
        except requests.RequestException as e:
            log.error("告知ステータス変更 失敗 : " + str(e))
            slack.notify(text="【ERROR】ConoHa 告知情報取得処理失敗しました。ログを確認してください。")
            sys.exit(1)

    # 未読の告知が0件の場合は処理を終了させる
    if not title_list:
        slack.notify(text="現在、未読の「ConoHaお知らせ」はありません。")
        log.info('ConoHa告知一覧取得 : 終了')
        sys.exit(0)

    slack.notify(text='\n'.join(title_list))
    log.info('ConoHa告知一覧取得 : 終了')

「get_notice.py」をcronで定時実行するようにすればConoHaのお知らせを定期的に確認しつつ、未読お知らせが溜まらなくなる仕組みの完成です。