kintoneで設定したスケジュールにあわせてlambda(python)からSQSへメッセージを送る
EC2の起動停止をそろそろ手動でやるのも疲れてきそうなのと、やはり停止するのを忘れてしまってもったいないので。
SQSを挟むほどの規模でもないのですが、慣れておきたかったのでSQSを使ってみています。
今回はSQSへメッセージを送るところまでを記載しています。
SQSからメッセージを受信してEC2の起動停止をlambdaで実行して結果をslackへ通知するのは次回のブログに書きます。
kintoneでアプリ作成
こんな項目を持ったアプリを作成します。
フォームはこんな感じです。
APIトークンを発行します
アプリの設定で詳細設定を展開して[APIトークン]をクリックします。
APIトークンの生成をします。
今回はデータを読むだけなのでアクセス権は閲覧のみにします。
保存した後、[設定完了]をクリックします。
※これをしないと生成したAPIトークンが反映しません。
ターミナルなどで確認します。
1 2 |
$ curl -H "X-Cybozu-API-Token: YOUR_TOKEN" "https://YOUR_DOMAIN.cybozu.com/k/v1/record.json?app=24&id=1" |
YOUR_TOKEN, YOUR_DOMAINは適宜変更します。
データのレスポンスが返ってくればOKです。
Terminalで見づらいのでPOSTMANで見ます。
祝日対応
やりながら祝日も対応しなければと思い、上記のkintoneの曜日に「祝」を足しました。
なんかいいAPIないかなーと探したのですが見つけられなかったのでGoogleカレンダーAPIを使いました。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
from apiclient.discovery import build API_KEY = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX' CALENDAR_ID = 'ja.japanese#holiday@group.v.calendar.google.com' def getHolidays(): try: service = build(serviceName='calendar', version='v3', developerKey=API_KEY) events = service.events().list(calendarId=CALENDAR_ID).execute() holidays = [] for item in sorted(events['items'], key=lambda item: item['start']['date']): holidays.append(item['start']['date']) return holidays except Exception as e: return e |
これをlambdaファンクションと同じディレクトリに格納しておきます。
kintoneからデータ取得してSQSへエンキューするlambda
GitHubはこちらです。
kintoneのレコードから起動対象か停止対象かをチェックします。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 |
def checkValidate(type_str,record): field_name = '' field_value = '' if type_str == 'start': field_name = 'c_start_check' field_value = u'開始' elif type_str == 'stop': field_name = 'c_stop_check' field_value = u'停止' check_item = record[field_name]['value'] if len(check_item) > 0: if check_item[0] == field_value: return True return False def checkDate(type_str,record,holidays,date_str,week_day_str): field_name = '' if type_str == 'start': field_name = 'c_start_week_day' elif type_str == 'stop': field_name = 'c_stop_week_day' week_day_array = record[field_name]['value'] if len(week_day_array) > 0: if u'祝' in week_day_array and date_str in holidays: return True elif week_day_str in week_day_array: if u'祝' in week_day_array: return True else: if date_str not in holidays: return True return False def checkItem(type_str,record,time_str): field_name = '' if type_str == 'start': field_name = 't_start_time' elif type_str == 'stop': field_name = 't_stop_time' check_time = record[field_name]['value'] if time_str[0:4] == check_time[0:4]: return True return False |
起動停止対象の場合辞書型のメッセージを配列に格納します。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
def getQueueList(): now = datetime.now(pytz.timezone('Asia/Tokyo')) date_str = now.strftime('%Y-%m-%d') time_str = now.strftime("%H:%M") week_day_str = getWeekDayStr(now) headers = {HEADERS_KEY:HEADERS_VALUE} response_record = requests.get(KINTONE_URL,headers=headers) record_data = json.loads(response_record.text) records=record_data['records'] holidays = get_holiday.getHolidays() queue_list = [] for record in records: ec2_name = record['s_ec2_name_tag']['value'] for type_str in ['start','stop']: if checkValidate(type_str,record): if checkDate(type_str,record,holidays,date_str,week_day_str): if checkItem(type_str,record,time_str): queue_list.append({'type':type_str,'ec2_name':ec2_name,'datetime':date_str + ' ' + time_str}) return queue_list |
SQSにメッセージをエンキューします。
1 2 3 4 5 6 7 8 9 10 11 |
def set_enqueue(queue_list): try: sqs_client = boto3.client('sqs') for queue in queue_list: response = sqs_client.send_message( QueueUrl = SQS_URL, MessageBody = json.dumps(queue) ) except Exception as e: logger.info(e) |
実行してます。
1 2 3 4 5 6 7 8 9 10 |
def handler(event, context): try: queue_list = getQueueList() if len(queue_list) > 0: set_enqueue(queue_list) except Exception as e: logger.info(e) raise e |
キューにメッセージが送信出来た事が確認出来ました。
次回はSQSからメッセージを受信してEC2を停止起動してslackへ通知します。

開発ベンダー5年、ユーザ企業システム部門通算9年、ITトレーナー2年目のSoftware Engineerです。
質問はコメントかSNSなどからお気軽にどうぞ。
出来る限りなるべく答えます。
このブログの内容/発言の一切は個人の見解であり、所属する組織とは関係ありません。
また、勉強会やイベントのレポートは自分が気になったことをメモしたり、聞いて思ったことを書いていますので、登壇者の意見や発表内容ではありません。
ad
ad
関連記事
-
-
テキストをAmazon PollyでMP3に変換してS3に格納(AWS Lambda Python)
Google Calendar Twilio ReminderのテキストをAma …
-
-
リザーブドインスタンスはじめました
このブログも2014年9月にはじめたので、もうすぐ3年。 1件~17件とばらつき …
-
-
AWS RDS でMySQLインスタンスを構築する
Amazon Web ServiceにはAmazon Relational Da …
-
-
API GatewayのアクセスログをCloudWatchに記録する
Amazon API GatewayのアクセスログをCloudWatchに記録で …
-
-
AWS Summit 2016 Tokyoに参加してきました (Day2)
馬込は非常に良い天気です。 泊まっている部屋が2Fでしたので窓を明けると歩いてい …
-
-
re:Invent 徒歩記(シアトルも)
このブログは、AWS re:invent 2018 Advent Calenda …
-
-
slackのbotにWikipediaを調べてもらう(Python on AWS Lambda + API Gateway)
slackのbotに少しでも役に立ってもらおうと、Wikipediaを調べてもら …
-
-
EC2 Auto Recovery機能を設定しておいた
以前EC2インスタンスのリタイア対象になったこともあり、というより、やっておいて …
-
-
AWS Lambda(Python3)でSelenium + Chrome Headless + でwebスクレイピングする
インターネット上に公開されている情報をDynamoDBにつっこみたいだけなので、 …
-
-
WordPressのwp-login.php , xmlrpc.phpへのアクセスをAWS WAFで接続元IPアドレスを制限する
AWS CloudWatch LogsエージェントでAmazon EC2上のNg …