はじめに: Boto3 とは？
Boto3 は、Amazon Web Services (AWS) の公式 Python SDK (Software Development Kit) です。Python 開発者が Amazon S3、Amazon EC2、Amazon DynamoDB などの様々な AWS サービスを利用するソフトウェアを簡単に開発できるように設計されています。Boto3 を使うことで、インフラのプロビジョニング、設定、管理、アプリケーションとの連携などを Python コードで自動化・効率化できます。
AWS のクラウドサービスを活用する上で、Boto3 は非常に強力なツールとなります。手動でのコンソール操作に比べて、以下のようなメリットがあります。
- 自動化: 定型的なタスクや複雑なワークフローをコードで自動化できます。
- 再現性: コードでインフラ構成を管理することで、同じ環境を何度でも正確に再現できます。
- 連携: Python アプリケーションやスクリプトと AWS サービスをシームレスに統合できます。
- スケーラビリティ: プログラムによる操作で、大量のリソース管理や大規模なデプロイメントに対応できます。
- 柔軟性: AWS の豊富なサービスを組み合わせ、高度なクラウドソリューションを構築できます。
このガイドでは、Boto3 の基本的な使い方から応用的なテクニック、主要な AWS サービスの操作例まで、幅広く解説していきます。さあ、Boto3 の世界へ飛び込みましょう！🚀
Boto3 のセットアップ 🛠️
インストール
Boto3 のインストールは、Python のパッケージ管理ツールである `pip` を使って簡単に行えます。ターミナルまたはコマンドプロンプトで以下のコマンドを実行してください。
pip install boto3依存関係のあるライブラリ (Botocore など) も自動的にインストールされます。プロジェクトごとに仮想環境を作成してインストールすることが推奨されます。
# 仮想環境を作成 (例: .venv)
python -m venv .venv
# 仮想環境をアクティベート (Linux/macOS)
source .venv/bin/activate
# 仮想環境をアクティベート (Windows)
.venv\Scripts\activate
# 仮想環境内で Boto3 をインストール
pip install boto3AWS 認証情報の設定
Boto3 が AWS サービスと通信するためには、AWS アカウントの認証情報が必要です。Boto3 は以下の順序で認証情報を検索し、最初に見つかったものを使用します。
- コード内での直接指定: `boto3.client()` や `boto3.Session()` の引数としてアクセスキーなどを渡す方法。 (セキュリティリスクが高いため非推奨)
- 環境変数: `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_SESSION_TOKEN` (オプション) などの環境変数を設定する方法。
- 共有認証情報ファイル: `~/.aws/credentials` ファイルに認証情報を記述する方法。複数のプロファイルを設定できます。
- AWS 設定ファイル: `~/.aws/config` ファイルに認証情報やリージョンなどを記述する方法。共有認証情報ファイルと組み合わせて使用できます。
- IAM ロール (Assume Role): `~/.aws/config` で指定されたプロファイルを使用して IAM ロールを引き受ける方法。Boto3 が自動的に STS の `AssumeRole` を呼び出します。
- Boto2 設定ファイル: 以前のバージョン (Boto2) の設定ファイル (`/etc/boto.cfg`, `~/.boto`)。
- EC2 インスタンスプロファイル / ECS コンテナ認証情報: IAM ロールがアタッチされた EC2 インスタンスや ECS タスク上で実行されている場合、インスタンスメタデータサービスやコンテナ認証情報プロバイダーから一時的な認証情報を自動的に取得します。これが最も推奨される方法です。
共有認証情報ファイル (`~/.aws/credentials`) の例
`aws configure` コマンド (AWS CLI の一部) を使うと、このファイルを簡単に作成・編集できます。
[default]
aws_access_key_id = YOUR_ACCESS_KEY_ID
aws_secret_access_key = YOUR_SECRET_ACCESS_KEY
[develop]
aws_access_key_id = ANOTHER_ACCESS_KEY_ID
aws_secret_access_key = ANOTHER_SECRET_ACCESS_KEYAWS 設定ファイル (`~/.aws/config`) の例
リージョンや出力形式、Assume Role の設定などを記述できます。
[default]
region = us-east-1
output = json
[profile develop]
region = ap-northeast-1
output = text
[profile cross_account_role]
role_arn = arn:aws:iam::123456789012:role/MyCrossAccountRole
source_profile = default
region = ap-northeast-1セキュリティの観点から、アクセスキーを直接コードや環境変数に埋め込むのではなく、IAM ロールや共有認証情報ファイルを使用することが強く推奨されます。特に EC2 や Lambda などの AWS 環境で実行する場合は、IAM ロールを使用するのがベストプラクティスです。
Boto3 の基本: クライアントとリソース 🤝
Boto3 には AWS サービスとやり取りするための主要なインターフェースとして、「クライアント (Client)」と「リソース (Resource)」の 2 種類があります。
クライアント (Client) – 低レベル API
- AWS サービスの API 操作と 1 対 1 でマッピングされています。
- サービスが提供する全ての API 操作をサポートしています。
- メソッド名は API 名をスネークケース (例: `ListBuckets` → `list_buckets`) にしたものです。
- レスポンスは通常、Python の辞書 (dict) 形式で返されます。
- より細かい制御が可能ですが、コードが冗長になる傾向があります。
- 内部的には Botocore ライブラリのクライアントオブジェクトを公開します。
- 一般的にスレッドセーフです (一部例外あり)。
import boto3
# S3 クライアントを作成
s3_client = boto3.client('s3')
# バケット一覧を取得
response = s3_client.list_buckets()
# バケット名を表示
print("Buckets:")
for bucket in response['Buckets']:
print(f" {bucket['Name']}")
# 特定のバケットのオブジェクト一覧を取得 (最大1000件)
try:
objects = s3_client.list_objects_v2(Bucket='your-bucket-name')
if 'Contents' in objects:
print("\nObjects in 'your-bucket-name':")
for obj in objects['Contents']:
print(f" - {obj['Key']} (Last Modified: {obj['LastModified']})")
else:
print("\nNo objects found in 'your-bucket-name'.")
except s3_client.exceptions.NoSuchBucket:
print("\nBucket 'your-bucket-name' does not exist.")
except Exception as e:
print(f"\nAn error occurred: {e}")
リソース (Resource) – 高レベル API
- AWS リソースをオブジェクト指向的に扱えるように抽象化されたインターフェースです。
- コードがより直感的で Pythonic になります。
- 属性へのアクセスやアクションの実行を通じて AWS リソースを操作します (例: `bucket.objects.all()`)。
- 明示的なネットワーク呼び出しを意識する必要が少なくなります。
- 注意: 全ての AWS サービスや全ての操作をサポートしているわけではありません。
- 重要 (2023年1月更新情報): AWS SDK for Python チームは、リソースインターフェースに新機能を追加しない方針を発表しました。既存のインターフェースは Boto3 のライフサイクル中は引き続き動作しますが、新しいサービス機能へのアクセスはクライアントインターフェースを通じて行われることになります。新規開発ではクライアント API の利用が推奨されます。
- スレッドセーフではありません。
import boto3
# S3 リソースを作成
s3_resource = boto3.resource('s3')
# バケット一覧を取得 (イテレータ)
print("Buckets:")
for bucket in s3_resource.buckets.all():
print(f" {bucket.name}")
# 特定のバケットオブジェクトを取得
try:
bucket = s3_resource.Bucket('your-bucket-name')
print("\nObjects in 'your-bucket-name':")
# オブジェクト一覧を取得 (イテレータ)
object_count = 0
for obj in bucket.objects.all():
print(f" - {obj.key} (Last Modified: {obj.last_modified})")
object_count += 1
if object_count == 0:
print("\nNo objects found in 'your-bucket-name'.")
except s3_resource.meta.client.exceptions.NoSuchBucket:
print("\nBucket 'your-bucket-name' does not exist.")
except Exception as e:
print(f"\nAn error occurred: {e}")
セッション (Session)
セッションは、認証情報や設定 (リージョンなど) を管理するためのコンテキストを提供します。`boto3.client()` や `boto3.resource()` を直接呼び出すと、暗黙的にデフォルトセッションが使用されます。
特定のプロファイルやリージョンを使いたい場合、あるいは設定をカスタマイズしたい場合は、明示的に `Session` オブジェクトを作成します。
import boto3
# 'develop' プロファイルと特定のリージョンを使用するセッションを作成
session = boto3.Session(profile_name='develop', region_name='ap-northeast-1')
# このセッションからクライアントやリソースを作成
ec2_client = session.client('ec2')
dynamodb_resource = session.resource('dynamodb')
# デフォルトセッションを使用する場合 (上記と等価ではない)
# default_ec2_client = boto3.client('ec2')
# default_dynamodb_resource = boto3.resource('dynamodb')
# セッションから認証情報を取得することも可能
credentials = session.get_credentials()
access_key = credentials.access_key
secret_key = credentials.secret_key
# print(f"Access Key: {access_key}") # セキュリティ上、実際のキーの出力は避けるべき
どちらを使うべきか？
以前は、簡単な操作にはリソース API、複雑な操作や未サポートの機能にはクライアント API という使い分けが一般的でした。しかし、AWS がリソース API の新規開発を停止したため、現在では新規プロジェクトにおいてはクライアント API の利用が一貫して推奨されます。
クライアント API は全ての AWS サービスと機能を網羅しており、長期的なメンテナンス性や将来性を考えるとより安全な選択肢となります。既存のコードでリソース API が使われている場合でも、徐々にクライアント API への移行を検討するのが良いでしょう。
主要な AWS サービス操作例 ⚙️
ここでは、いくつかの主要な AWS サービスについて、Boto3 (クライアント API) を使った基本的な操作例を紹介します。
Amazon S3 (Simple Storage Service) 📦
オブジェクトストレージサービスである S3 の操作例です。
バケットの作成・一覧表示・削除
import boto3
from botocore.exceptions import ClientError
s3_client = boto3.client('s3')
bucket_name = 'my-unique-boto3-test-bucket-12345' # グローバルに一意な名前が必要
region = boto3.session.Session().region_name # デフォルトリージョンを使用
# バケット作成 (リージョン指定が重要、特に us-east-1 以外)
try:
if region == 'us-east-1':
s3_client.create_bucket(Bucket=bucket_name)
else:
location = {'LocationConstraint': region}
s3_client.create_bucket(Bucket=bucket_name, CreateBucketConfiguration=location)
print(f"Bucket '{bucket_name}' created successfully in region '{region}'.")
except ClientError as e:
if e.response['Error']['Code'] == 'BucketAlreadyOwnedByYou':
print(f"Bucket '{bucket_name}' already exists and is owned by you.")
elif e.response['Error']['Code'] == 'BucketAlreadyExists':
print(f"Bucket '{bucket_name}' already exists and is owned by someone else.")
else:
print(f"Error creating bucket: {e}")
bucket_name = None # 作成失敗
# バケット一覧表示
try:
response = s3_client.list_buckets()
print("\nExisting buckets:")
for bucket in response['Buckets']:
print(f" - {bucket['Name']}")
except ClientError as e:
print(f"Error listing buckets: {e}")
# バケット削除 (バケットが空である必要がある)
if bucket_name:
try:
# バケットを空にする (本番環境では注意が必要！)
print(f"\nAttempting to empty bucket '{bucket_name}' before deletion...")
paginator = s3_client.get_paginator('list_objects_v2')
pages = paginator.paginate(Bucket=bucket_name)
objects_to_delete = []
for page in pages:
if 'Contents' in page:
for obj in page['Contents']:
objects_to_delete.append({'Key': obj['Key']})
if objects_to_delete:
# 一度に最大1000オブジェクトまで削除可能
for i in range(0, len(objects_to_delete), 1000):
chunk = objects_to_delete[i:i + 1000]
s3_client.delete_objects(Bucket=bucket_name, Delete={'Objects': chunk})
print(f"Emptied bucket '{bucket_name}'.")
else:
print(f"Bucket '{bucket_name}' is already empty.")
# バケット削除実行
s3_client.delete_bucket(Bucket=bucket_name)
print(f"Bucket '{bucket_name}' deleted successfully.")
except ClientError as e:
print(f"Error deleting bucket '{bucket_name}': {e}")
オブジェクトのアップロード・ダウンロード・削除
import boto3
from botocore.exceptions import ClientError
import os
s3_client = boto3.client('s3')
bucket_name = 'your-existing-bucket-name' # 事前に作成したバケット名
local_file_path = 'my_local_file.txt'
s3_object_key = 'uploads/my_remote_file.txt'
download_path = 'downloaded_file.txt'
# ローカルにダミーファイル作成
with open(local_file_path, 'w') as f:
f.write('This is a test file for Boto3 S3 upload.\nHello, AWS! 👋')
print(f"Created dummy file: '{local_file_path}'")
# オブジェクトのアップロード (ファイルから)
try:
s3_client.upload_file(local_file_path, bucket_name, s3_object_key)
print(f"Successfully uploaded '{local_file_path}' to 's3://{bucket_name}/{s3_object_key}'")
except ClientError as e:
print(f"Error uploading file: {e}")
except FileNotFoundError:
print(f"Error: Local file '{local_file_path}' not found.")
# オブジェクトのダウンロード
try:
print(f"\nDownloading 's3://{bucket_name}/{s3_object_key}' to '{download_path}'...")
s3_client.download_file(bucket_name, s3_object_key, download_path)
print(f"Successfully downloaded to '{download_path}'.")
# ダウンロードしたファイルの内容を表示 (確認用)
with open(download_path, 'r') as f:
print("Downloaded file content:")
print(f.read())
except ClientError as e:
if e.response['Error']['Code'] == "404":
print(f"Error: Object 's3://{bucket_name}/{s3_object_key}' not found.")
else:
print(f"Error downloading file: {e}")
# オブジェクトの削除
try:
print(f"\nDeleting object 's3://{bucket_name}/{s3_object_key}'...")
s3_client.delete_object(Bucket=bucket_name, Key=s3_object_key)
print("Successfully deleted the object.")
except ClientError as e:
print(f"Error deleting object: {e}")
# ローカルファイルのクリーンアップ
if os.path.exists(local_file_path):
os.remove(local_file_path)
print(f"\nRemoved local file: '{local_file_path}'")
if os.path.exists(download_path):
os.remove(download_path)
print(f"Removed downloaded file: '{download_path}'")
upload_file や download_file は、大きなファイルの場合に自動的にマルチパートアップロード/ダウンロードを行ってくれる便利な高レベルメソッドです。
署名付き URL の生成
プライベートな S3 オブジェクトに対して、一時的なアクセス権を付与する URL を生成できます。
import boto3
from botocore.exceptions import ClientError
s3_client = boto3.client('s3')
bucket_name = 'your-private-bucket-name' # プライベートなバケット名
object_key = 'private/document.pdf'
expiration_seconds = 3600 # URL の有効期間 (秒単位、ここでは1時間)
# GET リクエスト用の署名付き URL を生成
try:
signed_url = s3_client.generate_presigned_url(
'get_object',
Params={'Bucket': bucket_name, 'Key': object_key},
ExpiresIn=expiration_seconds
)
print(f"Generated presigned URL (valid for {expiration_seconds} seconds):")
print(signed_url)
print("\nNote: This URL grants temporary access. Keep it secure.")
except ClientError as e:
print(f"Error generating presigned URL: {e}")
# PUT リクエスト用の署名付き URL を生成 (アップロード用)
try:
signed_url_put = s3_client.generate_presigned_url(
'put_object',
Params={'Bucket': bucket_name, 'Key': 'uploads/new_file.zip'},
ExpiresIn=expiration_seconds,
HttpMethod='PUT' # PUTの場合は指定が必要な場合あり
)
print(f"\nGenerated presigned URL for PUT (valid for {expiration_seconds} seconds):")
print(signed_url_put)
print("\nUse this URL with a PUT request (e.g., curl -T file.zip 'URL') to upload.")
except ClientError as e:
print(f"Error generating presigned URL for PUT: {e}")
Amazon EC2 (Elastic Compute Cloud) 🖥️
仮想サーバー (インスタンス) を管理する EC2 の操作例です。
インスタンスの起動・情報取得・停止・終了
import boto3
from botocore.exceptions import ClientError
import time
ec2_client = boto3.client('ec2')
# AMI ID はリージョンやOSによって異なります。適切なものを選択してください。
# 例: Amazon Linux 2 AMI (HVM) - SSD Volume Type in us-east-1 (これは古くなっている可能性あり)
# 最新の AMI ID は AWS マネジメントコンソールや AWS CLI で確認してください。
ami_id = 'ami-xxxxxxxxxxxxxxxxx' # ここに適切なAMI IDを入力
instance_type = 't2.micro' # 無料利用枠の対象 (変更可能)
key_name = 'your-key-pair-name' # 事前に作成したキーペア名
security_group_ids = ['sg-xxxxxxxxxxxxxxxxx'] # 事前に作成したセキュリティグループID
instance_id = None
# インスタンスの起動
try:
print(f"Launching an EC2 instance ({instance_type}) using AMI '{ami_id}'...")
response = ec2_client.run_instances(
ImageId=ami_id,
InstanceType=instance_type,
KeyName=key_name,
SecurityGroupIds=security_group_ids,
MinCount=1,
MaxCount=1,
TagSpecifications=[ # インスタンスにタグを付ける
{
'ResourceType': 'instance',
'Tags': [
{'Key': 'Name', 'Value': 'Boto3-Launched-Instance'},
{'Key': 'Project', 'Value': 'Boto3-Guide'},
]
},
]
)
instance_id = response['Instances'][0]['InstanceId']
print(f"Instance launch requested. Instance ID: {instance_id}")
# インスタンスが 'running' 状態になるまで待機 (Waiterを使用)
print("Waiting for the instance to reach 'running' state...")
waiter = ec2_client.get_waiter('instance_running')
waiter.wait(InstanceIds=[instance_id])
print("Instance is now running!")
except ClientError as e:
print(f"Error launching instance: {e}")
except Exception as e:
print(f"An unexpected error occurred during launch: {e}")
# インスタンス情報の取得
if instance_id:
try:
print(f"\nDescribing instance {instance_id}...")
response = ec2_client.describe_instances(InstanceIds=[instance_id])
instance_info = response['Reservations'][0]['Instances'][0]
instance_state = instance_info['State']['Name']
public_ip = instance_info.get('PublicIpAddress', 'N/A')
private_ip = instance_info.get('PrivateIpAddress', 'N/A')
tags = instance_info.get('Tags', [])
print(f" State: {instance_state}")
print(f" Public IP: {public_ip}")
print(f" Private IP: {private_ip}")
print(" Tags:")
for tag in tags:
print(f" {tag['Key']}: {tag['Value']}")
except ClientError as e:
print(f"Error describing instance: {e}")
# インスタンスの停止
try:
print(f"\nStopping instance {instance_id}...")
ec2_client.stop_instances(InstanceIds=[instance_id])
# インスタンスが 'stopped' 状態になるまで待機
print("Waiting for the instance to reach 'stopped' state...")
waiter = ec2_client.get_waiter('instance_stopped')
waiter.wait(InstanceIds=[instance_id])
print("Instance is now stopped.")
# 状態を再確認
response = ec2_client.describe_instances(InstanceIds=[instance_id])
instance_state = response['Reservations'][0]['Instances'][0]['State']['Name']
print(f" Current State: {instance_state}")
except ClientError as e:
print(f"Error stopping instance: {e}")
# 少し待機 (API制限を避けるため)
time.sleep(10)
# インスタンスの終了 (削除)
try:
print(f"\nTerminating instance {instance_id}...")
ec2_client.terminate_instances(InstanceIds=[instance_id])
# インスタンスが 'terminated' 状態になるまで待機
print("Waiting for the instance to reach 'terminated' state...")
waiter = ec2_client.get_waiter('instance_terminated')
waiter.wait(InstanceIds=[instance_id])
print("Instance terminated successfully.")
except ClientError as e:
# すでに終了している場合のエラーコードを確認する必要があるかも
print(f"Error terminating instance: {e}")
else:
print("\nInstance was not launched due to previous errors.")
注意: 上記コードはインスタンスを起動し、その後終了させます。EC2 インスタンスには料金が発生するため、テスト後は不要なインスタンスが残っていないか確認してください。
Amazon DynamoDB (NoSQL Database) 💾
高速でスケーラブルな NoSQL データベースサービスである DynamoDB の操作例です。
テーブルの作成・アイテムの追加・取得・更新・削除・クエリ
import boto3
from botocore.exceptions import ClientError
from decimal import Decimal # DynamoDB の数値型 (Number) は Decimal を使う
import time
dynamodb_client = boto3.client('dynamodb')
table_name = 'Boto3-Movies-Table'
# テーブル作成
try:
print(f"Creating DynamoDB table '{table_name}'...")
response = dynamodb_client.create_table(
TableName=table_name,
AttributeDefinitions=[
{'AttributeName': 'year', 'AttributeType': 'N'}, # N: Number
{'AttributeName': 'title', 'AttributeType': 'S'}, # S: String
],
KeySchema=[
{'AttributeName': 'year', 'KeyType': 'HASH'}, # パーティションキー
{'AttributeName': 'title', 'KeyType': 'RANGE'}, # ソートキー
],
ProvisionedThroughput={ # オンデマンドキャパシティの場合は不要
'ReadCapacityUnits': 1,
'WriteCapacityUnits': 1
}
# BillingMode='PAY_PER_REQUEST' # オンデマンドの場合
)
print("Table creation requested. Waiting for table to become active...")
# テーブルがアクティブになるまで待機 (Waiterを使用)
waiter = dynamodb_client.get_waiter('table_exists')
waiter.wait(TableName=table_name)
print(f"Table '{table_name}' is now active.")
except ClientError as e:
if e.response['Error']['Code'] == 'ResourceInUseException':
print(f"Table '{table_name}' already exists.")
else:
print(f"Error creating table: {e}")
table_name = None # 作成失敗
if table_name:
# アイテムの追加 (PutItem)
try:
print("\nAdding items to the table...")
items_to_add = [
{'year': {'N': '2013'}, 'title': {'S': 'Rush'}, 'info': {'M': {'plot': {'S': 'Formula 1 drama'}, 'rating': {'N': '8.1'}}}},
{'year': {'N': '2014'}, 'title': {'S': 'Interstellar'}, 'info': {'M': {'plot': {'S': 'Space exploration'}, 'rating': {'N': '8.6'}}}},
{'year': {'N': '2014'}, 'title': {'S': 'Whiplash'}, 'info': {'M': {'plot': {'S': 'Intense drumming'}, 'rating': {'N': '8.5'}}}}
]
for item in items_to_add:
dynamodb_client.put_item(TableName=table_name, Item=item)
print(f" Added: Year={item['year']['N']}, Title='{item['title']['S']}'")
print("Items added successfully.")
except ClientError as e:
print(f"Error adding items: {e}")
# アイテムの取得 (GetItem)
try:
print("\nGetting item (Year=2014, Title='Interstellar')...")
response = dynamodb_client.get_item(
TableName=table_name,
Key={'year': {'N': '2014'}, 'title': {'S': 'Interstellar'}}
)
if 'Item' in response:
item = response['Item']
# Decimal 型に注意して表示
rating = Decimal(item['info']['M']['rating']['N'])
print(f" Found item: {item['title']['S']} ({item['year']['N']}) - Rating: {rating}")
print(f" Plot: {item['info']['M']['plot']['S']}")
else:
print(" Item not found.")
except ClientError as e:
print(f"Error getting item: {e}")
# アイテムの更新 (UpdateItem) - 評価 (rating) を更新し、新しい属性を追加
try:
print("\nUpdating item (Year=2014, Title='Interstellar') - setting rating to 9.0 and adding 'director'...")
response = dynamodb_client.update_item(
TableName=table_name,
Key={'year': {'N': '2014'}, 'title': {'S': 'Interstellar'}},
UpdateExpression="SET info.rating = :r, info.director = :d", # 属性を更新/追加
ExpressionAttributeValues={
':r': {'N': '9.0'}, # 新しい評価
':d': {'S': 'Christopher Nolan'} # 新しい属性 (監督)
},
ReturnValues="UPDATED_NEW" # 更新後の属性値を返す
)
print(" Item updated successfully.")
print(" Updated attributes:", response['Attributes'])
except ClientError as e:
print(f"Error updating item: {e}")
# クエリ (Query) - 特定の年 (2014) の映画を取得
try:
print("\nQuerying for movies released in 2014...")
response = dynamodb_client.query(
TableName=table_name,
KeyConditionExpression="year = :y", # パーティションキー 'year' が特定の値 (:y) と一致
ExpressionAttributeValues={
":y": {"N": "2014"}
}
)
print(f" Found {len(response['Items'])} movies from 2014:")
for item in response['Items']:
rating = Decimal(item['info']['M']['rating']['N'])
director = item['info']['M'].get('director', {}).get('S', 'N/A') # 追加した属性
print(f" - {item['title']['S']} (Rating: {rating}, Director: {director})")
except ClientError as e:
print(f"Error querying items: {e}")
# アイテムの削除 (DeleteItem)
try:
print("\nDeleting item (Year=2013, Title='Rush')...")
dynamodb_client.delete_item(
TableName=table_name,
Key={'year': {'N': '2013'}, 'title': {'S': 'Rush'}}
)
print(" Item deleted successfully.")
except ClientError as e:
print(f"Error deleting item: {e}")
# スキャン (Scan) - テーブル全体をスキャン (注意: 大規模テーブルでは非効率)
# 本番環境では Scan は避け、Query やインデックスを検討すべき
try:
print("\nScanning the entire table (use with caution on large tables)...")
response = dynamodb_client.scan(TableName=table_name)
print(f" Found {len(response['Items'])} items in total after deletion:")
for item in response['Items']:
print(f" - {item['title']['S']} ({item['year']['N']})")
except ClientError as e:
print(f"Error scanning table: {e}")
# テーブル削除 (テスト後など)
delete_table = input(f"\nDo you want to delete the table '{table_name}'? (yes/no): ").lower()
if delete_table == 'yes':
try:
print(f"Deleting table '{table_name}'...")
dynamodb_client.delete_table(TableName=table_name)
print("Waiting for table to be deleted...")
waiter = dynamodb_client.get_waiter('table_not_exists')
waiter.wait(TableName=table_name)
print(f"Table '{table_name}' deleted successfully.")
except ClientError as e:
print(f"Error deleting table: {e}")
else:
print(f"Table '{table_name}' was not deleted.")
else:
print("\nTable operations skipped due to creation error.")
DynamoDB では、属性の型 (S: String, N: Number, B: Binary, M: Map, L: List など) を明示的に指定する必要があります。また、数値型には Python の `Decimal` 型を使用します。
AWS Lambda (Serverless Compute) λ
サーバーレスコンピューティングサービスである Lambda 関数の操作例です。
関数の呼び出し (Invoke)
既存の Lambda 関数を同期または非同期で呼び出します。
import boto3
import json
from botocore.exceptions import ClientError
lambda_client = boto3.client('lambda')
function_name = 'my-existing-lambda-function-name' # 呼び出す Lambda 関数の名前または ARN
payload = {
'key1': 'value1',
'key2': 'value2',
'number': 123
}
# 同期呼び出し (結果を待つ)
try:
print(f"Invoking Lambda function '{function_name}' synchronously...")
response = lambda_client.invoke(
FunctionName=function_name,
InvocationType='RequestResponse', # 同期呼び出し
Payload=json.dumps(payload) # ペイロードは JSON 文字列に変換
)
status_code = response['StatusCode']
print(f" Invocation Status Code: {status_code}")
if 'FunctionError' in response:
print(f" Function Error: {response['FunctionError']}")
# エラー時のペイロード (詳細)
error_payload = json.loads(response['Payload'].read().decode('utf-8'))
print(" Error Payload:", json.dumps(error_payload, indent=2))
else:
# 成功時のペイロード (関数の戻り値)
result_payload = json.loads(response['Payload'].read().decode('utf-8'))
print(" Function Result Payload:", json.dumps(result_payload, indent=2))
except ClientError as e:
if e.response['Error']['Code'] == 'ResourceNotFoundException':
print(f"Error: Lambda function '{function_name}' not found.")
else:
print(f"Error invoking Lambda function synchronously: {e}")
except Exception as e:
print(f"An unexpected error occurred during synchronous invocation: {e}")
# 非同期呼び出し (結果を待たない)
try:
print(f"\nInvoking Lambda function '{function_name}' asynchronously...")
response = lambda_client.invoke(
FunctionName=function_name,
InvocationType='Event', # 非同期呼び出し
Payload=json.dumps(payload)
)
status_code = response['StatusCode']
# 非同期の場合、通常は 202 Accepted が返る (リクエスト受け付け成功)
print(f" Invocation Request Status Code: {status_code}")
if status_code == 202:
print(" Asynchronous invocation request accepted.")
else:
print(" Asynchronous invocation request might have failed.")
# 非同期呼び出しでは、関数の実行結果はこのレスポンスには含まれない
except ClientError as e:
if e.response['Error']['Code'] == 'ResourceNotFoundException':
print(f"Error: Lambda function '{function_name}' not found.")
else:
print(f"Error invoking Lambda function asynchronously: {e}")
except Exception as e:
print(f"An unexpected error occurred during asynchronous invocation: {e}")
Lambda 関数の作成、更新、削除、パーミッション設定なども Boto3 で行うことができますが、デプロイパッケージ (ZIP ファイル) の準備などが必要になるため、ここでは呼び出しの例に留めます。関数の管理には AWS SAM CLI や Serverless Framework などのツールを利用することも一般的です。
Boto3 の応用テクニック ✨
Boto3 には、AWS サービスとのやり取りをより効率的かつ堅牢にするための便利な機能がいくつか用意されています。
ページネーション (Paginators) 📖
`list_objects_v2` (S3) や `describe_instances` (EC2) のように、一度のリクエストで取得できる結果の数に上限がある API 操作があります。これらの API から全ての結果を取得するには、レスポンスに含まれる次のページのトークンを使って繰り返しリクエストを送る必要があります。
Paginator はこの繰り返し処理を抽象化し、簡単に全結果を反復処理できるようにします。
import boto3
from botocore.exceptions import ClientError
s3_client = boto3.client('s3')
bucket_name = 'your-bucket-with-many-objects'
try:
print(f"Listing all objects in bucket '{bucket_name}' using Paginator...")
paginator = s3_client.get_paginator('list_objects_v2')
# ページごとに処理する場合
# page_iterator = paginator.paginate(Bucket=bucket_name)
# for page_num, page in enumerate(page_iterator):
# print(f"--- Page {page_num + 1} ---")
# if 'Contents' in page:
# for obj in page['Contents']:
# print(f" - {obj['Key']} (Size: {obj['Size']})")
# else:
# print(" (No objects on this page)")
# 全てのオブジェクトをフラットに処理する場合 (メモリに注意)
object_count = 0
operation_parameters = {'Bucket': bucket_name}
page_iterator = paginator.paginate(**operation_parameters)
for page in page_iterator:
if 'Contents' in page:
for obj in page['Contents']:
# ここで各オブジェクトに対する処理を行う
# print(f"Processing object: {obj['Key']}")
object_count += 1
print(f"\nFinished processing. Total objects found: {object_count}")
except ClientError as e:
if e.response['Error']['Code'] == 'NoSuchBucket':
print(f"Error: Bucket '{bucket_name}' does not exist.")
else:
print(f"An error occurred: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
ウェイター (Waiters) ⏳
EC2 インスタンスの起動 (`run_instances`) や DynamoDB テーブルの作成 (`create_table`) など、一部の AWS 操作は非同期的に実行され、リソースが目的の状態になるまで時間がかかります。
Waiter は、特定のリソースが指定された状態 (例: `instance_running`, `table_exists`, `bucket_exists`) に達するまで、自動的にポーリングを行い待機する機能を提供します。これにより、`time.sleep()` を使った自前のポーリングロジックを実装する必要がなくなります。
import boto3
from botocore.exceptions import ClientError, WaiterError
ec2_client = boto3.client('ec2')
instance_id = 'i-xxxxxxxxxxxxxxxxx' # 起動直後などのインスタンス ID
try:
print(f"Waiting for instance {instance_id} to be running...")
waiter = ec2_client.get_waiter('instance_running')
# wait() メソッドで待機開始
# デフォルトでは、一定回数リトライしても状態に達しない場合 WaiterError が発生
waiter.wait(
InstanceIds=[instance_id],
WaiterConfig={
'Delay': 15, # ポーリング間隔 (秒) - デフォルトは 15
'MaxAttempts': 40 # 最大リトライ回数 - デフォルトは 40
}
)
print(f"Instance {instance_id} is now running!")
except WaiterError as e:
print(f"Waiter timed out or failed for instance {instance_id}: {e}")
# エラーの詳細 (最後のレスポンスなど) は e.last_response で確認可能
except ClientError as e:
# describe_instances などで発生する可能性のあるエラー
if e.response['Error']['Code'] == 'InvalidInstanceID.NotFound':
print(f"Error: Instance ID '{instance_id}' not found.")
else:
print(f"A ClientError occurred: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
# --- DynamoDB テーブルが存在するまで待つ例 ---
dynamodb_client = boto3.client('dynamodb')
table_name = 'my-newly-created-table'
try:
print(f"\nWaiting for DynamoDB table '{table_name}' to exist and be active...")
waiter = dynamodb_client.get_waiter('table_exists')
waiter.wait(TableName=table_name)
print(f"Table '{table_name}' is active.")
except WaiterError as e:
print(f"Waiter timed out or failed for table '{table_name}': {e}")
except ClientError as e:
print(f"A ClientError occurred while waiting for the table: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
エラーハンドリング (Error Handling) 🚨
Boto3 を使って AWS サービスと通信する際には、様々なエラーが発生する可能性があります (ネットワークエラー、認証エラー、API 制限超過、リソースが見つからない、パラメータが不正など)。堅牢なアプリケーションを構築するには、これらのエラーを適切に処理することが不可欠です。
Boto3 (および内部で使われている Botocore) が発生させる例外は主に以下の 2 種類です。
- `botocore.exceptions.ClientError`: AWS サービスからのエラーレスポンス (HTTP ステータスコードが 4xx または 5xx) を表します。最も一般的に遭遇するエラーです。エラーの詳細 (エラーコード、メッセージなど) はレスポンスの `Error` ディクショナリに含まれます。
- その他の `botocore.exceptions`: クライアント側の設定ミス (`ConfigNotFound`, `NoCredentialsError`) や接続の問題 (`EndpointConnectionError`, `ConnectTimeoutError`) など、AWS サービスからのレスポンスではないエラーを表します。
`ClientError` をキャッチし、その中の `response[‘Error’][‘Code’]` を確認することで、具体的なエラー内容に応じた処理を行うことができます。
import boto3
from botocore.exceptions import ClientError, NoCredentialsError, PartialCredentialsError, EndpointConnectionError
s3_client = boto3.client('s3')
bucket_name = 'non-existent-bucket-qwertyuiop'
object_key = 'some_object.txt'
try:
print(f"Attempting to get object 's3://{bucket_name}/{object_key}'...")
response = s3_client.get_object(Bucket=bucket_name, Key=object_key)
# 成功した場合の処理 ...
print("Object retrieved successfully (this likely won't be reached).")
except ClientError as e:
error_code = e.response['Error']['Code']
error_message = e.response['Error']['Message']
request_id = e.response['ResponseMetadata']['RequestId']
http_status_code = e.response['ResponseMetadata']['HTTPStatusCode']
print(f"\nCaught a ClientError:")
print(f" Error Code: {error_code}")
print(f" Message: {error_message}")
print(f" Request ID: {request_id}")
print(f" HTTP Status Code: {http_status_code}")
if error_code == 'NoSuchBucket':
print(" Handling: The specified bucket does not exist.")
# バケット作成処理や代替処理を行うなど
elif error_code == 'NoSuchKey':
print(" Handling: The specified object key does not exist in the bucket.")
# オブジェクト作成処理やデフォルト値を使うなど
elif error_code in ('ProvisionedThroughputExceededException', 'ThrottlingException'):
print(" Handling: Request throttled. Consider implementing backoff and retry.")
# time.sleep( exponentially increasing delay ) など
elif error_code == 'AccessDenied':
print(" Handling: Access denied. Check IAM permissions for the credentials.")
else:
print(" Handling: An unexpected AWS service error occurred.")
# 必要に応じてエラーを再 raise するか、一般的なエラー処理を行う
# raise e
except (NoCredentialsError, PartialCredentialsError) as e:
print(f"\nCaught a Credentials Error: {e}")
print(" Handling: AWS credentials not found or incomplete. Check configuration.")
# 環境変数、認証情報ファイル、IAMロールなどを確認
except EndpointConnectionError as e:
print(f"\nCaught an Endpoint Connection Error: {e}")
print(" Handling: Could not connect to the AWS endpoint. Check network connectivity and region.")
except Exception as e:
# 予期しないその他の例外
print(f"\nCaught an unexpected exception: {type(e).__name__} - {e}")
# 必要に応じてログ記録やエラー通知を行う
raise e # 再 raise して上位で処理させることも可能
具体的なエラーコードは、各 AWS サービスのドキュメントや API リファレンスで確認できます。適切なエラーハンドリングを行うことで、アプリケーションの信頼性と安定性が向上します。
セッションとクライアント/リソースのカスタマイズ 🔧
`boto3.Session` やクライアント/リソースを作成する際に、様々なオプションを指定して動作をカスタマイズできます。
- `region_name`: 使用する AWS リージョンを指定します。
- `profile_name`: `~/.aws/credentials` や `~/.aws/config` で定義された特定のプロファイルを使用します。
- `aws_access_key_id`, `aws_secret_access_key`, `aws_session_token`: 認証情報を直接指定します (非推奨)。
- `config`: `botocore.config.Config` オブジェクトを指定して、リトライ戦略、タイムアウト、プロキシ設定などを細かく制御します。
- `endpoint_url`: 標準の AWS エンドポイント以外 (例: ローカル開発用の LocalStack や VPC エンドポイント) を指定します。
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
# カスタムリトライ設定と接続タイムアウトを設定した Config オブジェクト
custom_config = Config(
region_name='ap-southeast-2', # シドニーリージョン
signature_version='v4', # 署名バージョン
retries={
'max_attempts': 5, # 最大リトライ回数
'mode': 'adaptive' # リトライモード (standard, legacy, adaptive)
},
connect_timeout=10, # 接続タイムアウト (秒)
read_timeout=30, # 読み取りタイムアウト (秒)
# http_proxy='http://proxy.example.com:8080', # プロキシ設定例
# https_proxy='https://proxy.example.com:8080',
)
# 'staging' プロファイルとカスタム設定でセッションを作成
try:
session = boto3.Session(profile_name='staging', config=custom_config)
# このセッションからクライアントを作成 (設定が引き継がれる)
s3_client_custom = session.client('s3')
dynamodb_client_custom = session.client('dynamodb', endpoint_url='http://localhost:8000') # Local DynamoDB を指定
print("Session and clients created with custom configuration:")
print(f" Region: {s3_client_custom.meta.region_name}")
print(f" Endpoint (DynamoDB): {dynamodb_client_custom.meta.endpoint_url}")
print(f" Retry Max Attempts: {s3_client_custom.meta.config.retries['max_attempts']}")
# カスタム設定を使って S3 バケットをリスト (存在しないプロファイルや接続できない場合はエラー)
response = s3_client_custom.list_buckets()
print("\nBuckets (using custom config):")
for bucket in response['Buckets']:
print(f" - {bucket['Name']}")
except ClientError as e:
print(f"\nAn error occurred with custom config client: {e}")
except Exception as e:
print(f"\nAn unexpected error occurred: {e}")
# --- リージョンだけを直接指定してクライアントを作成 ---
try:
ec2_tokyo = boto3.client('ec2', region_name='ap-northeast-1')
print(f"\nCreated EC2 client for Tokyo region: {ec2_tokyo.meta.region_name}")
# 東京リージョンの情報を取得など
# response = ec2_tokyo.describe_regions()
# print(response)
except Exception as e:
print(f"\nError creating EC2 client for Tokyo: {e}")
Boto3 のベストプラクティス 👍
Boto3 を効果的かつ安全に使用するためのベストプラクティスをいくつか紹介します。
-
クライアント API を優先する:
前述の通り、リソース API の新規開発は停止されています。新規開発や既存コードの改修においては、将来性や網羅性の観点からクライアント API (`boto3.client()`) の利用を一貫して推奨します。 -
認証情報を安全に管理する:
アクセスキーをコードにハードコーディングしたり、バージョン管理システムにコミットしたりしないでください。IAM ロール (EC2/ECS/Lambda 環境) や、共有認証情報ファイル (`~/.aws/credentials`)、環境変数 (一時的な利用) など、安全な方法で認証情報を管理しましょう。最小権限の原則に従い、必要な権限のみを付与した IAM ポリシーを使用してください。 -
適切なエラーハンドリングを実装する:
`try…except` ブロックを使用して `ClientError` やその他の潜在的な例外をキャッチし、適切に処理します。特定のエラーコード (`response[‘Error’][‘Code’]`) に基づいて、リトライ、代替処理、ユーザーへの通知などを行います。予期しないエラーも考慮し、ログ記録や監視を行いましょう。 -
リトライメカニズムを理解し、活用する:
Boto3 (Botocore) にはデフォルトでリトライ機能が組み込まれています (一時的なネットワークエラーやスロットリング (`ThrottlingException`) などに対して)。`botocore.config.Config` を使ってリトライ回数やモード (standard, adaptive) を調整できます。ただし、リトライすべきでないエラー (例: `AccessDenied`, `ValidationException`) もあるため、エラーコードに応じて適切に対応する必要があります。 -
Paginator と Waiter を活用する:
大量の結果を処理する場合は Paginator を、リソースの状態変化を待つ場合は Waiter を利用して、コードをシンプルかつ効率的に保ちましょう。自前のポーリングやページネーションロジックは複雑になりがちです。 -
最新バージョンを利用する:
Boto3 と Botocore は定期的に更新され、新しい AWS サービスのサポート、バグ修正、パフォーマンス改善が行われます。`pip install –upgrade boto3 botocore` で定期的に最新バージョンにアップデートすることを検討してください。 -
セッションを効果的に利用する:
複数のリージョンや異なる認証情報プロファイルを使用する場合、あるいはリトライ設定などをカスタマイズしたい場合は、明示的に `boto3.Session` オブジェクトを作成して利用します。これにより、設定の管理が容易になります。 -
コストを意識する:
API コールにはコストが発生する場合があります (特に高頻度な呼び出しやデータ転送)。不要な API 呼び出しを避け、効率的なクエリ (例: DynamoDB の Scan ではなく Query) を使用し、必要に応じて結果をキャッシュするなどの工夫をしましょう。特にループ内で大量の API を呼び出す場合は注意が必要です。 -
ドキュメントを参照する:
Boto3 の公式ドキュメントは非常に充実しています。各サービスのクライアントメソッド、パラメータ、レスポンス形式、利用可能な Waiter や Paginator など、詳細な情報が記載されています。困ったときはまずドキュメントを確認しましょう。
まとめ 🎉
Boto3 は、Python を使って AWS の強力なクラウドサービス群を自在に操るための必須ライブラリです。インフラの自動化、アプリケーションとの連携、データ処理、サーバーレス開発など、その活用範囲は非常に広大です。
このガイドでは、Boto3 の基本的なセットアップから、クライアント API を中心とした主要サービスの操作、そして Paginator, Waiter, エラーハンドリングといった応用的なテクニック、さらにはベストプラクティスまでを解説しました。
Boto3 を使いこなすことで、AWS 環境における開発・運用の効率と信頼性を大幅に向上させることができます。ぜひ、実際のプロジェクトで Boto3 を活用し、クラウドの可能性を最大限に引き出してください！ Happy coding! 😊