[614]python库boto3

pip install boto3

import boto3

# 创建S3资源对象
s3 = boto3.resource('s3')

# 创建一个新的S3桶
bucket = s3.create_bucket(Bucket='my-new-bucket')

# 上传文件
s3.Object('my-new-bucket', 'my-file.txt').put(Body=open('my-file.txt', 'rb'))

# 列出桶里的对象
for obj in bucket.objects.all():
    print(obj.key)

import boto3

# 创建EC2资源对象
ec2 = boto3.resource('ec2')

# 启动新的EC2实例
instances = ec2.create_instances(
    ImageId='ami-0abcdef1234567890',
    MinCount=1,
    MaxCount=1,
    InstanceType='t2.micro'
)

print(instances[0].id)

import boto3

# 创建DynamoDB资源对象
dynamodb = boto3.resource('dynamodb')

# 创建表
table = dynamodb.create_table(
    TableName='MyTable',
    KeySchema=[
        {
            'AttributeName': 'username',
            'KeyType': 'HASH'  # 分区键
        },
        {
            'AttributeName': 'last_name',
            'KeyType': 'RANGE'  # 排序键
        }
    ],
    AttributeDefinitions=[
        {
            'AttributeName': 'username',
            'AttributeType': 'S'
        },
        {
            'AttributeName': 'last_name',
            'AttributeType': 'S'
        },
    ],
    ProvisionedThroughput={
        'ReadCapacityUnits': 10,
        'WriteCapacityUnits': 10
    }
)

# 插入数据
table.put_item(
   Item={
        'username': 'janedoe',
        'last_name': 'Doe',
        'age': 25,
        'email': 'janedoe@example.com'
    }
)

# 查询数据
response = table.get_item(
    Key={
        'username': 'janedoe',
        'last_name': 'Doe'
    }
)
print(response['Item'])

import boto3

# 创建S3客户端
s3 = boto3.client('s3')

# 创建分页器
paginator = s3.get_paginator('list_objects_v2')

# 使用分页器
page_iterator = paginator.paginate(Bucket='my-bucket')

# 遍历分页器中的每一页
for page in page_iterator:
    # 输出当前页中的对象键
    for obj in page['Contents']:
        print(obj['Key'])

import boto3
import logging

# 创建S3客户端
s3 = boto3.client('s3')

# 定义一个回调函数来处理事件
def log_request_params(event_name, params, **kwargs):
    logging.info(f"Before call: {event_name} with params {params}")

def log_response(**kwargs):
    logging.info(f"After call: {kwargs}")

# 订阅事件
s3.meta.events.register('before-call.s3', log_request_params)
s3.meta.events.register('after-call.s3', log_response)

# 执行一个S3操作，比如列出桶内容
s3.list_buckets()

import boto3
from botocore.config import Config

# 创建一个自定义配置
custom_config = Config(
    region_name='us-west-2',
    retries={
        'max_attempts': 10,
        'mode': 'standard'
    }
)

# 使用自定义配置创建S3客户端
s3 = boto3.client('s3', config=custom_config)

# 使用这个定制的客户端进行操作
response = s3.list_buckets()
print(response)