Zenn
💻

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

2024/12/24に公開
AWS
Python
Boto3
tech

概要

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

前提

  • Python version: 3.12.4
  • AWS CLIを触ったことがあること

注意

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

すみ分け

IaCとの違い

IaCは主にインフラストラクチャ全体の管理や一貫性のあるリソース構成の際に利用します。
対してSDKは、動的なリソース操作や複数サービス間の連携に強みを持ち、特に複雑なロジックや条件に基づくリソース管理に便利です。

CLIとの違い

CLIは主に単体のサービスに対して操作します。
簡単な操作や一時的な変更には便利ですが、複雑なロジックやリソース間の依存関係を持つ操作には向いていません。
SDKは複数サービス間の連携や、返り値を元にした動的な処理に強みがあります。

ユースケース

  • リソース単体の設定値を確認したい
    ⇒ CLI
  • 複数リソースを一括で定義して構築したい
    ⇒ IaC
  • 複数の環境（開発、ステージング、本番）で同一のインフラを構築したい
    ⇒ IaC
  • 一連の複数リソース情報を一括で取得したい
    ⇒ SDK or IaC
    ※リソースをIaCで一括管理している場合は、例えばterraform.stateファイルを元に解析したほうが良い場合もあります。
  • DB等から取得したデータを元にリソースを動的操作したい
    ⇒ SDK

基本操作

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

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を利用してメトリクスフィルターを全件取得します。

ケース別使用例

複数のリソース情報を取得し結合する

例として、インスタンス名とVPC名のリストを作成しlist_instance_vpcname.jsonファイルに保存してみます。

get_instance_name_vpcname_pairs.py
import boto3
import json

client = boto3.client('ec2')

path_output = './list_instance_vpcname.json' # 保存先パス

def get_instance_name_vpcid_pairs():
  """ 
    インスタンス名とVPC IDの配列を作成します。
    [{'instance_name': 'インスタンス タグ名 ※1', 'vpc_id': 'VPC ID'}]
    ※1 タグ名が無い場合はIDが入ります。
  """
  response = client.describe_instances()
  instance_vpcid_pairs = []
  for r in response['Reservations']:
    for i in r['Instances']:
      instance_vpcid_pairs.append({
        'instance_name': next((tag['Value'] for tag in i.get('Tags', []) if tag['Key'] == 'Name'), i['InstanceId']),
        'vpc_id': i['VpcId']
      })
  return instance_vpcid_pairs

def get_vpcid_to_vpcname():
  """ 
    VPC IDとVPC名の配列を作成します。
    [{'VPC ID': 'VPC 名 ※1'}]
    ※1 タグ名が無い場合はIDが入ります。
  """
  response = client.describe_vpcs() # VPC数が多い場合はNextToken処理を入れる
  vpcid_vpcname_pairs = []
  for v in response['Vpcs']:
    vpcid_vpcname_pairs.append({
      v['VpcId']: next((tag['Value'] for tag in v.get('Tags', []) if tag['Key'] == 'Name'), v['VpcId'])
    })
  return vpcid_vpcname_pairs

def get_instance_name_vpcname_pairs(instance_vpcid_pairs, vpcid_to_vpcname):
  """ 
  インスタンス名とVPC名の配列を作成します。
  [{'instance_name': 'インスタンス名', 'vpc_name': 'VPC名'}]
  """
  instance_vpcname_pairs = []
  for i in instance_vpcid_pairs:
    instance_vpcname_pairs.append({
      'instance_name': i['instance_name'],
      'vpc_name': next((vpc[i['vpc_id']] for vpc in vpcid_to_vpcname if i['vpc_id'] in vpc))
    })
  return instance_vpcname_pairs

def main():
  instance_vpcid_pairs = get_instance_name_vpcid_pairs()
  vpcid_to_vpcname = get_vpcid_to_vpcname()
  instance_vpcname_pairs = get_instance_name_vpcname_pairs(instance_vpcid_pairs, vpcid_to_vpcname)
  with open(path_output, 'w') as f:
    print(json.dumps(instance_vpcname_pairs, indent=2), file=f)
  return

if __name__ == '__main__':
  main()

実行してみましょう。

python get_instance_name_vpcname_pairs.py
---
[
  {
    "instance_name": "hoge-instance",
    "vpc_name": "hoge-vpc"
  }
  :(省略)
]
---

動的にリソースを操作する

例として、タグ{'StartTime': '0800'}がついているEC2インスタンスを起動させてみましょう。
唐突ですが、今回はloggingを利用してログ出力をしてみます。
loggingの設定をスクリプトごとに記述するのは面倒なので別ファイルに定義します。

setup_logger.py
import logging

def setup_logger(level=logging.INFO):
  """ 
  logging 設定

  Parameters
  ----------
  level : str, default INFO
    ログレベル e.g. DEBUG, INFO, ERROR

  Returns
  -------
  logging.Loger
    設定されたロガーインスタンス
  """
  # logger 作成
  logger = logging.getLogger(__name__)
  logger.setLevel(level)
  # コンソール出力用ハンドラ作成
  console_handler = logging.StreamHandler()
  # 出力フォーマット設定
  formatter = logging.Formatter('%(asctime)s\t[%(levelname)s]\t%(message)s')
  console_handler.setFormatter(formatter)
  # ハンドラを logger に追加
  logger.addHandler(console_handler)
  
  return logger

以下のスクリプトは、特定のタグを持つEC2インスタンスを起動します。
setup_logger.pyで定義したsetup_loggerを利用してログ出力します。

start_instances_by_tag.py
""" 
処理内容：特定のタグを持つEC2インスタンスを起動します。
実行方法：python start_instances_by_tag.py <tag_key> <tag_value>

Parameters
----------
tag_key : str
  タグのキー
tag_value : str
  タグの値

Returns
-------
None
"""
import boto3
import sys
import setup_logger

logger = setup_logger.setup_logger()

client = boto3.client('ec2')

def get_instances_by_tag(tag_key, tag_value):
  """ 
  指定されたタグを持つEC2インスタンスを取得する。

  Parameters
  ----------
  tag_key : str
    検索するタグのキー
  tag_value : str
    検索するタグの値

  Returns
  -------
  list_instance_ids : list
    指定されたタグを持つEC2インスタンスのIDリスト
  """
  try:
    response = client.describe_instances(Filters=[
      {'Name': f'tag:{tag_key}', 'Values': [tag_value]}
    ])
    
    list_instance_ids = []
    for r in response['Reservations']:
      for i in r['Instances']:
        list_instance_ids.append(i['InstanceId'])
    return list_instance_ids
  except Exception as e:
    logger.error('インスタンスリスト作成中にエラーが発生しました。')
    raise

def start_instances_by_tag(list_instance_ids):
  """ 
  指定されたインスタンスIDを持つEC2インスタンスを起動する。

  Parameters
  ----------
  list_instance_ids : list
    起動するインスタンスIDリスト

  Returns
  -------
  None
  """
  if not list_instance_ids:
    logger.info('起動するインスタンスが見つかりませんでした。')
    return
  
  try:
    for i in list_instance_ids:
      logger.info(f'\t- {i} を起動します。')
      client.start_instances(InstanceIds=[i])
    return
  except Exception as e:
    logger.error('インスタンス起動処理にてエラーが発生しました。')
    raise

def main():
  # 引数確認
  if len(sys.argv) != 3:
    logger.error('引数が不正です。')
    logger.error('実行方法： `python start_instances_by_tag.py <tag_key> <tag_value>`')
    sys.exit(1)
  tag_key = sys.argv[1]
  tag_value = sys.argv[2]
  
  logger.info('インスタンス起動処理を開始します。')
  logger.info(f'tag_key: {tag_key}, tag_value: {tag_value}')
  list_instance_ids = get_instances_by_tag(tag_key, tag_value)
  start_instances_by_tag(list_instance_ids)
  logger.info('インスタンス起動処理が完了しました。')

if __name__ == "__main__":
  main()

実行してみましょう。

python start_instances_by_tag.py AutoStart 0800
---
yyyy-MM-dd hh:mm:ss [INFO]  インスタンス起動処理を開始します。
yyyy-MM-dd hh:mm:ss [INFO]  tag_key: AutoStart, tag_value: 0800
yyyy-MM-dd hh:mm:ss [INFO]          - i-XXXXXXXXXXXXXXXXX を起動します。
yyyy-MM-dd hh:mm:ss [INFO]          - i-XXXXXXXXXXXXXXXXX を起動します。
yyyy-MM-dd hh:mm:ss [INFO]          - i-XXXXXXXXXXXXXXXXX を起動します。
yyyy-MM-dd hh:mm:ss [INFO]  インスタンス起動処理が完了しました。
---

Discussion