💻

【インフラエンジニア向け】AWS SDK for Python（boto3）事始め

に公開
AWS
Python
Boto3
tech

概要

インフラエンジニア向けの、AWS SDK for Python (boto3) 導入記事です。

前提

  • Python version: 3.12.4
  • Boto3 version: 3.16.*

注意

  • Python, pip, AWS CLIのインストールや設定については触れません。
  • 完全に独学なので変なことしてたら教えてほしいです。

基本操作

本項では、基本的な使い方を紹介します。

Clientインスタンスの生成

import boto3

client = boto3.client('ec2')
response = client.describe_instances()

print(response)

上記のコードは、EC2インスタンスの情報を取得します。CLIの aws ec2 describe-instances と同じ動作です。
詳しく解説していきます。

import boto3
client = boto3.client('ec2')

boto3ライブラリを読み込み、AWSサービスに対応するClientクラスのインスタンスオブジェクトを生成します。
本コードではEC2クライアントを作成します。
boto3からAWSサービスを操作する際には必ず記述するコードなので覚えておきましょう。

response = client.describe_instances()

作成したEC2クライアント（クラスインスタンス）を利用して describe_instances APIを発行し、戻り値を response へ格納します。

print(response)

response を出力します。

NextToken の利用

boto3 には1度のAPIで取得できるリソース数に上限があるものがあります。
取得上限を超えて情報取得する際にNextTokenを利用します。

import boto3

client = boto3.client('logs')

_describe_metric_filters = []
next_token = None
while True:
  if next_token:
    response = client.describe_metric_filters(nextToken=next_token)
  else:
    response = client.describe_metric_filters()
  _describe_metric_filters.extend(response['metricFilters'])
  next_token = response.get('nextToken')
  if not next_token:
    break
print(_describe_metric_filters)

上記のコードは、NextTokenを利用してメトリクスフィルターを全件取得します。

コード例

任意のタグが付いているAlarmのステータスを無効化する機能を試しに書いてみます。

ソースコード

おおまかに処理の流れを説明します。

  1. describe_alarms_by_alarm_typeでアラーム一覧を取得します。
    デフォルトでメトリックアラームを取得します。
  2. 1で取得したアラーム情報を元にlist_tags_for_resourceでアラームのタグを取得します。
  3. 指定したタグをもつアラーム名のリストを作成します。
  4. 3で作成したリストのアラームをenable_alarm_actionsで有効化します。
enable_alarms_by_tag.py
import logging
import json
import boto3

# logging setting
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

class CloudWatchClient:
  def __init__(self):
    self.client = boto3.client('cloudwatch')
  
  def _check_response(self, response):
    """ API 呼び出しのHTTPステータスコードを確認する """
    status_code = response['ResponseMetadata']['HTTPStatusCode']
    if status_code != 200:
      logger.error(f'API呼び出しのHTTPステータスコードが不正です。HTTPステータスコード: {status_code}')
      raise
    return
  
  def describe_alarms_by_alarm_type(self, alarm_type: str = 'MetricAlarm'):
    """ API describe_alarms """
    try:
      _response_describe_alarms = []
      next_token = None

      while True:
        if next_token:
          response = self.client.describe_alarms(AlarmTypes = [alarm_type], NextToken = next_token)
        else:
          response = self.client.describe_alarms(AlarmTypes = [alarm_type])

        self._check_response(response)
        _response_describe_alarms += response[alarm_type + 's']

        next_token = response.get('NextToken', None)
        if not next_token:
          break

      return _response_describe_alarms
    except Exception as e:
      logger.error('アラーム情報取得処理中にエラーが発生しました。')
      raise
  
  def list_tags_for_resource(self, resource_arn: str):
    """ API list_tags_for_resource """
    try:
      response = self.client.list_tags_for_resource(ResourceARN = resource_arn)
      self._check_response(response)
      return response
    except Exception as e:
      logger.error('アラームのタグ情報取得処理中にエラーが発生しました。')
      raise

  def enable_alarm_actions(self, alarm_names: list):
    """ API enable_alarm_actions """
    try:
      response = self.client.enable_alarm_actions(AlarmNames = alarm_names)
      self._check_response(response)
      return response
    except Exception as e:
      logger.error('アラーム有効化処理中にエラーが発生しました。')
      raise

def main(tag_key: str, tag_value: str, alarm_type: str = 'MetricAlarm'):
  """ 
  指定したタグを持つアラームを有効化します。
  
  Parameters
  ----------
  tag_key : str
  tag_value : str
  
  Returns
  -------
  None
  """
  # CloudWatch クライアントインスタンス作成
  cloudwatch_client = CloudWatchClient()
  
  # アラーム情報一覧を取得する
  response_describe_alarms = cloudwatch_client.describe_alarms_by_alarm_type(alarm_type)
  
  # 指定したタグを持つアラーム名を取得する
  target_alarm_names = []
  for alarm in response_describe_alarms:
    response_tags = cloudwatch_client.list_tags_for_resource(alarm['AlarmArn'])
    if any(tag['Key'] == tag_key and tag['Value'] == tag_value for tag in response_tags['Tags']):
      target_alarm_names.append(alarm['AlarmName'])
  
  # 指定したタグを持つアラームを有効化する
  cloudwatch_client.enable_alarm_actions(target_alarm_names)
  logger.info(f'以下のアラームを有効化しました。\n{json.dumps(target_alarm_names, default=str, indent=2)}')
  return

実行してみる

ファイル末尾に追記します。
main()を呼ぶ際にタグを指定します。

enable_alarms_by_tag.py
if __name__ == '__main__':
  params = {
    'tag_key': 'favorite_animal',
    'tag_value': 'gorilla'
  }
  main(**params)

実行します。

$ python enable_alarms_by_tag.py
以下のアラームを有効化しました。
[
  'test-alarm-01',
  'test-alarm-02',
  :(省略)
]

さいごに

2種類以上のAPIを利用する処理においては、Bash + AWS CLIなどで頑張るよりこっちのほうが良いと考えます。

Discussion