【存储工程】S3 API 深度解析

# 最终一致性时代的典型补偿模式（已不再需要）
import time

def put_and_verify(s3_client, bucket, key, body, max_retries=5):
    """上传后轮询验证——在强一致性时代已无必要"""
    s3_client.put_object(Bucket=bucket, Key=key, Body=body)

    for attempt in range(max_retries):
        try:
            response = s3_client.head_object(Bucket=bucket, Key=key)
            return response
        except s3_client.exceptions.ClientError:
            time.sleep(2 ** attempt)  # 指数退避

    raise RuntimeError(f"对象 {key} 在 {max_retries} 次重试后仍不可见")

<?xml version="1.0" encoding="UTF-8"?>
<Error>
  <Code>NoSuchKey</Code>
  <Message>The specified key does not exist.</Message>
  <Key>my-missing-object.txt</Key>
  <RequestId>tx00000000000000000001-006...</RequestId>
  <HostId>...</HostId>
</Error>

import boto3

s3 = boto3.client('s3')

# 模拟 "列出 photos/2025/ 下的子目录和文件"
response = s3.list_objects_v2(
    Bucket='my-data',
    Prefix='photos/2025/',
    Delimiter='/'
)

# CommonPrefixes 包含"子目录"
for prefix in response.get('CommonPrefixes', []):
    print(f"目录: {prefix['Prefix']}")
    # 输出:
    # 目录: photos/2025/01/
    # 目录: photos/2025/02/

# Contents 包含当前"目录"下的直接对象
for obj in response.get('Contents', []):
    print(f"文件: {obj['Key']}, 大小: {obj['Size']}")

import hashlib
import hmac
import datetime


def sign(key, msg):
    """HMAC-SHA256 签名"""
    return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()


def get_signature_key(secret_key, date_stamp, region, service):
    """派生签名密钥"""
    k_date = sign(('AWS4' + secret_key).encode('utf-8'), date_stamp)
    k_region = sign(k_date, region)
    k_service = sign(k_region, service)
    k_signing = sign(k_service, 'aws4_request')
    return k_signing


def create_authorization_header(
    method, uri, query_string, headers,
    payload, access_key, secret_key, region, service
):
    """构建完整的 Authorization 头部"""
    t = datetime.datetime.utcnow()
    amz_date = t.strftime('%Y%m%dT%H%M%SZ')
    date_stamp = t.strftime('%Y%m%d')

    # 步骤 1：规范请求
    canonical_headers = ''
    signed_headers_list = sorted(headers.keys())
    for h in signed_headers_list:
        canonical_headers += f'{h}:{headers[h]}\n'
    signed_headers = ';'.join(signed_headers_list)

    payload_hash = hashlib.sha256(payload.encode('utf-8')).hexdigest()

    canonical_request = '\n'.join([
        method, uri, query_string,
        canonical_headers, signed_headers, payload_hash
    ])

    # 步骤 2：待签名字符串
    credential_scope = f'{date_stamp}/{region}/{service}/aws4_request'
    string_to_sign = '\n'.join([
        'AWS4-HMAC-SHA256', amz_date,
        credential_scope,
        hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()
    ])

    # 步骤 3-4：计算签名
    signing_key = get_signature_key(secret_key, date_stamp, region, service)
    signature = hmac.new(
        signing_key, string_to_sign.encode('utf-8'), hashlib.sha256
    ).hexdigest()

    # 构建 Authorization 头部
    authorization = (
        f'AWS4-HMAC-SHA256 '
        f'Credential={access_key}/{credential_scope}, '
        f'SignedHeaders={signed_headers}, '
        f'Signature={signature}'
    )
    return authorization

import boto3
import hashlib
import base64

s3 = boto3.client('s3', region_name='us-east-1')

# 基础上传
s3.put_object(
    Bucket='my-bucket',
    Key='data/report.json',
    Body=b'{"status": "ok"}',
    ContentType='application/json',
    # 服务端加密
    ServerSideEncryption='AES256',
    # 用户自定义元数据
    Metadata={
        'author': 'ltl',
        'version': '1.0'
    },
    # 存储类别
    StorageClass='STANDARD',
    # 设置缓存控制（通过 CloudFront 分发时有用）
    CacheControl='max-age=86400',
)

# 带条件写入（防止覆盖已有对象）
try:
    s3.put_object(
        Bucket='my-bucket',
        Key='config/settings.json',
        Body=b'{"debug": false}',
        # 仅当对象不存在时才写入
        IfNoneMatch='*',
    )
except s3.exceptions.ClientError as e:
    if e.response['Error']['Code'] == 'PreconditionFailed':
        print("对象已存在，跳过写入")

# 带内容校验的上传
data = b'important data that must not be corrupted'
md5_hash = base64.b64encode(hashlib.md5(data).digest()).decode()

s3.put_object(
    Bucket='my-bucket',
    Key='critical/data.bin',
    Body=data,
    ContentMD5=md5_hash,  # S3 会验证完整性
)

import boto3

s3 = boto3.client('s3', region_name='us-east-1')

# 基础下载
response = s3.get_object(Bucket='my-bucket', Key='data/report.json')
body = response['Body'].read()
content_type = response['ContentType']
last_modified = response['LastModified']

# 范围请求（Range Request）—— 下载部分内容
response = s3.get_object(
    Bucket='my-bucket',
    Key='large-file.bin',
    Range='bytes=0-1023'  # 只下载前 1KB
)
partial_data = response['Body'].read()

# 条件请求 —— 避免不必要的传输
response = s3.get_object(
    Bucket='my-bucket',
    Key='data/report.json',
    IfModifiedSince=last_modified,  # 仅当修改后才返回
)

# 流式读取大文件（避免一次性加载到内存）
response = s3.get_object(Bucket='my-bucket', Key='huge-file.csv')
stream = response['Body']

chunk_size = 8 * 1024 * 1024  # 8MB
while True:
    chunk = stream.read(chunk_size)
    if not chunk:
        break
    process_chunk(chunk)

stream.close()

import boto3

s3 = boto3.client('s3', region_name='us-east-1')


def list_all_objects(bucket, prefix=''):
    """使用分页器列出所有对象"""
    paginator = s3.get_paginator('list_objects_v2')
    page_iterator = paginator.paginate(
        Bucket=bucket,
        Prefix=prefix,
        PaginationConfig={
            'PageSize': 1000,  # 每页最多 1000
        }
    )

    total_count = 0
    total_size = 0

    for page in page_iterator:
        for obj in page.get('Contents', []):
            total_count += 1
            total_size += obj['Size']
            yield obj

    print(f"共 {total_count} 个对象，总大小 {total_size / (1024**3):.2f} GB")


# 使用生成器逐个处理
for obj in list_all_objects('my-bucket', prefix='logs/2025/'):
    if obj['Key'].endswith('.gz'):
        print(f"{obj['Key']} -> {obj['Size']} bytes")

import boto3

s3 = boto3.client('s3', region_name='us-east-1')


def batch_delete(bucket, prefix, dry_run=True):
    """批量删除指定前缀下的所有对象"""
    paginator = s3.get_paginator('list_objects_v2')
    pages = paginator.paginate(Bucket=bucket, Prefix=prefix)

    delete_count = 0

    for page in pages:
        objects = page.get('Contents', [])
        if not objects:
            continue

        if dry_run:
            for obj in objects:
                print(f"[DRY RUN] 将删除: {obj['Key']}")
            delete_count += len(objects)
            continue

        # 每次最多删除 1000 个对象
        delete_list = [{'Key': obj['Key']} for obj in objects]
        response = s3.delete_objects(
            Bucket=bucket,
            Delete={
                'Objects': delete_list,
                'Quiet': True,  # 静默模式，只返回错误
            }
        )

        errors = response.get('Errors', [])
        if errors:
            for err in errors:
                print(f"删除失败: {err['Key']} -> {err['Code']}")

        delete_count += len(objects) - len(errors)

    print(f"{'[DRY RUN] ' if dry_run else ''}共删除 {delete_count} 个对象")

import os
import hashlib
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import boto3
from botocore.config import Config


class MultipartUploader:
    """生产级分片上传器，支持并发上传与断点续传"""

    def __init__(self, bucket, region='us-east-1', max_concurrency=10):
        self.bucket = bucket
        self.s3 = boto3.client(
            's3',
            region_name=region,
            config=Config(
                max_pool_connections=max_concurrency + 5,
                retries={'max_attempts': 3, 'mode': 'adaptive'},
            )
        )
        self.max_concurrency = max_concurrency

    def upload_file(self, file_path, key, part_size=64 * 1024 * 1024):
        """上传文件，自动选择单次上传或分片上传"""
        file_size = os.path.getsize(file_path)

        # 小于 100MB 使用单次上传
        if file_size < 100 * 1024 * 1024:
            self._simple_upload(file_path, key)
            return

        self._multipart_upload(file_path, key, part_size)

    def _simple_upload(self, file_path, key):
        """单次上传"""
        with open(file_path, 'rb') as f:
            self.s3.put_object(Bucket=self.bucket, Key=key, Body=f)
        print(f"单次上传完成: {key}")

    def _multipart_upload(self, file_path, key, part_size):
        """分片上传"""
        file_size = os.path.getsize(file_path)

        # 步骤 1：初始化分片上传
        response = self.s3.create_multipart_upload(
            Bucket=self.bucket,
            Key=key,
            ContentType=self._guess_content_type(key),
        )
        upload_id = response['UploadId']
        print(f"分片上传已初始化: UploadId={upload_id}")

        try:
            # 步骤 2：并发上传分片
            parts = self._upload_parts(
                file_path, key, upload_id, file_size, part_size
            )

            # 步骤 3：完成分片上传
            self.s3.complete_multipart_upload(
                Bucket=self.bucket,
                Key=key,
                UploadId=upload_id,
                MultipartUpload={
                    'Parts': sorted(parts, key=lambda x: x['PartNumber'])
                },
            )
            print(f"分片上传完成: {key} ({file_size / (1024**3):.2f} GB)")

        except Exception as e:
            # 上传失败时中止，释放已上传的分片（避免产生费用）
            print(f"上传失败，中止分片上传: {e}")
            self.s3.abort_multipart_upload(
                Bucket=self.bucket, Key=key, UploadId=upload_id
            )
            raise

    def _upload_parts(self, file_path, key, upload_id, file_size, part_size):
        """并发上传所有分片"""
        parts = []
        lock = threading.Lock()
        uploaded_bytes = 0

        def upload_single_part(part_number, offset, size):
            nonlocal uploaded_bytes
            with open(file_path, 'rb') as f:
                f.seek(offset)
                data = f.read(size)

            response = self.s3.upload_part(
                Bucket=self.bucket,
                Key=key,
                UploadId=upload_id,
                PartNumber=part_number,
                Body=data,
            )

            with lock:
                uploaded_bytes += size
                progress = (uploaded_bytes / file_size) * 100
                print(f"  分片 {part_number} 完成 ({progress:.1f}%)")

            return {
                'PartNumber': part_number,
                'ETag': response['ETag'],
            }

        # 构建分片任务列表
        tasks = []
        offset = 0
        part_number = 1
        while offset < file_size:
            size = min(part_size, file_size - offset)
            tasks.append((part_number, offset, size))
            offset += size
            part_number += 1

        # 并发执行
        with ThreadPoolExecutor(max_workers=self.max_concurrency) as executor:
            futures = {
                executor.submit(upload_single_part, pn, off, sz): pn
                for pn, off, sz in tasks
            }
            for future in as_completed(futures):
                result = future.result()
                parts.append(result)

        return parts

    @staticmethod
    def _guess_content_type(key):
        """根据文件扩展名猜测 MIME 类型"""
        import mimetypes
        content_type, _ = mimetypes.guess_type(key)
        return content_type or 'application/octet-stream'


# 使用示例
if __name__ == '__main__':
    uploader = MultipartUploader(
        bucket='my-data-bucket',
        region='us-east-1',
        max_concurrency=8,
    )
    uploader.upload_file(
        file_path='/data/large-dataset.parquet',
        key='datasets/2025/large-dataset.parquet',
        part_size=128 * 1024 * 1024,  # 128MB 分片
    )

import boto3


def resume_upload(s3_client, bucket, key, upload_id, file_path, part_size):
    """恢复中断的分片上传"""
    # 查询已上传的分片
    uploaded_parts = {}
    paginator = s3_client.get_paginator('list_parts')
    for page in paginator.paginate(
        Bucket=bucket, Key=key, UploadId=upload_id
    ):
        for part in page.get('Parts', []):
            uploaded_parts[part['PartNumber']] = part['ETag']

    print(f"已上传 {len(uploaded_parts)} 个分片，从断点继续...")

    file_size = os.path.getsize(file_path)
    all_parts = []
    offset = 0
    part_number = 1

    while offset < file_size:
        size = min(part_size, file_size - offset)

        if part_number in uploaded_parts:
            # 该分片已上传，跳过
            all_parts.append({
                'PartNumber': part_number,
                'ETag': uploaded_parts[part_number],
            })
            print(f"  分片 {part_number} 已存在，跳过")
        else:
            # 上传缺失的分片
            with open(file_path, 'rb') as f:
                f.seek(offset)
                data = f.read(size)

            response = s3_client.upload_part(
                Bucket=bucket, Key=key,
                UploadId=upload_id,
                PartNumber=part_number,
                Body=data,
            )
            all_parts.append({
                'PartNumber': part_number,
                'ETag': response['ETag'],
            })
            print(f"  分片 {part_number} 上传完成")

        offset += size
        part_number += 1

    # 完成上传
    s3_client.complete_multipart_upload(
        Bucket=bucket, Key=key, UploadId=upload_id,
        MultipartUpload={
            'Parts': sorted(all_parts, key=lambda x: x['PartNumber'])
        },
    )
    print("断点续传完成")

import boto3
from datetime import datetime, timezone, timedelta


def cleanup_incomplete_uploads(s3_client, bucket, max_age_days=7):
    """清理超过指定天数的未完成分片上传"""
    cutoff = datetime.now(timezone.utc) - timedelta(days=max_age_days)

    paginator = s3_client.get_paginator('list_multipart_uploads')
    cleaned = 0

    for page in paginator.paginate(Bucket=bucket):
        for upload in page.get('Uploads', []):
            if upload['Initiated'] < cutoff:
                s3_client.abort_multipart_upload(
                    Bucket=bucket,
                    Key=upload['Key'],
                    UploadId=upload['UploadId'],
                )
                print(f"已中止: {upload['Key']} (UploadId: {upload['UploadId']})")
                cleaned += 1

    print(f"共清理 {cleaned} 个未完成的分片上传")


# 更好的做法：配置 Bucket 生命周期策略自动清理
lifecycle_rule = {
    'Rules': [{
        'ID': 'abort-incomplete-multipart',
        'Status': 'Enabled',
        'Filter': {'Prefix': ''},
        'AbortIncompleteMultipartUpload': {
            'DaysAfterInitiation': 7
        }
    }]
}

import boto3
import json

s3 = boto3.client('s3', region_name='us-east-1')


def query_csv_with_s3_select(bucket, key, sql_expression):
    """使用 S3 Select 查询 CSV 文件"""
    response = s3.select_object_content(
        Bucket=bucket,
        Key=key,
        ExpressionType='SQL',
        Expression=sql_expression,
        InputSerialization={
            'CSV': {
                'FileHeaderInfo': 'USE',  # 使用第一行作为列名
                'RecordDelimiter': '\n',
                'FieldDelimiter': ',',
                'QuoteCharacter': '"',
            },
            'CompressionType': 'GZIP',
        },
        OutputSerialization={
            'JSON': {
                'RecordDelimiter': '\n',
            },
        },
    )

    records = []
    for event in response['Payload']:
        if 'Records' in event:
            payload = event['Records']['Payload'].decode('utf-8')
            for line in payload.strip().split('\n'):
                if line:
                    records.append(json.loads(line))
        elif 'Stats' in event:
            stats = event['Stats']['Details']
            print(f"扫描字节数: {stats['BytesScanned']:,}")
            print(f"处理字节数: {stats['BytesProcessed']:,}")
            print(f"返回字节数: {stats['BytesReturned']:,}")
            ratio = stats['BytesReturned'] / stats['BytesScanned'] * 100
            print(f"数据缩减比: {ratio:.2f}%")

    return records


# 查询示例：从访问日志中筛选错误请求
results = query_csv_with_s3_select(
    bucket='my-logs',
    key='access-logs/2025/09/25.csv.gz',
    sql_expression="""
        SELECT s.timestamp, s.method, s.path, s.status_code, s.response_time
        FROM s3object s
        WHERE CAST(s.status_code AS INT) >= 500
        AND CAST(s.response_time AS FLOAT) > 1.0
        LIMIT 100
    """,
)

for r in results:
    print(f"{r['timestamp']} {r['method']} {r['path']} -> {r['status_code']}")


def query_parquet_with_s3_select(bucket, key):
    """查询 Parquet 文件——Parquet 的列式特性使 S3 Select 更高效"""
    response = s3.select_object_content(
        Bucket=bucket,
        Key=key,
        ExpressionType='SQL',
        Expression="""
            SELECT user_id, event_type, COUNT(*) as cnt
            FROM s3object s
            WHERE event_type = 'purchase'
        """,
        InputSerialization={
            'Parquet': {},  # Parquet 不需要额外配置
        },
        OutputSerialization={
            'JSON': {'RecordDelimiter': '\n'},
        },
    )

    for event in response['Payload']:
        if 'Records' in event:
            print(event['Records']['Payload'].decode('utf-8'))

import boto3

s3 = boto3.client('s3', region_name='us-east-1')

# Glacier Select 通过 RestoreObject 发起
s3.restore_object(
    Bucket='my-archive',
    Key='historical-data/2020-transactions.csv.gz',
    RestoreRequest={
        'Type': 'SELECT',
        'Tier': 'Standard',  # Standard: 3-5 小时; Expedited: 1-5 分钟
        'SelectParameters': {
            'InputSerialization': {
                'CSV': {'FileHeaderInfo': 'USE'},
                'CompressionType': 'GZIP',
            },
            'ExpressionType': 'SQL',
            'Expression': """
                SELECT * FROM s3object s
                WHERE s.amount > '10000'
                AND s.currency = 'USD'
            """,
            'OutputSerialization': {
                'CSV': {},
            },
        },
        'OutputLocation': {
            'S3': {
                'BucketName': 'my-results',
                'Prefix': 'glacier-select-output/',
            },
        },
    },
)

import boto3

s3 = boto3.client('s3', region_name='us-east-1')

# 启用版本控制
s3.put_bucket_versioning(
    Bucket='my-bucket',
    VersioningConfiguration={'Status': 'Enabled'},
)

# 上传同一个 Key 的多个版本
s3.put_object(Bucket='my-bucket', Key='config.json', Body=b'{"v": 1}')
s3.put_object(Bucket='my-bucket', Key='config.json', Body=b'{"v": 2}')
s3.put_object(Bucket='my-bucket', Key='config.json', Body=b'{"v": 3}')

# 列出所有版本
response = s3.list_object_versions(Bucket='my-bucket', Prefix='config.json')
for version in response.get('Versions', []):
    print(
        f"VersionId: {version['VersionId']}, "
        f"IsLatest: {version['IsLatest']}, "
        f"LastModified: {version['LastModified']}, "
        f"Size: {version['Size']}"
    )

# 获取特定版本
response = s3.get_object(
    Bucket='my-bucket',
    Key='config.json',
    VersionId='specific-version-id-here',
)
old_content = response['Body'].read()

# 删除操作在启用版本控制后的行为：
# 不指定 VersionId 的 DELETE -> 插入删除标记（Delete Marker）
# 指定 VersionId 的 DELETE -> 永久删除该版本
s3.delete_object(Bucket='my-bucket', Key='config.json')
# 此时 GET config.json 返回 404，但所有历史版本仍然存在

# 恢复被删除的对象：删除"删除标记"
response = s3.list_object_versions(Bucket='my-bucket', Prefix='config.json')
for marker in response.get('DeleteMarkers', []):
    if marker['IsLatest']:
        s3.delete_object(
            Bucket='my-bucket',
            Key='config.json',
            VersionId=marker['VersionId'],
        )
        print("已恢复对象（删除了删除标记）")
        break

import boto3
from datetime import datetime, timezone, timedelta

s3 = boto3.client('s3', region_name='us-east-1')

# 创建启用对象锁定的 Bucket（只能在创建时启用）
s3.create_bucket(
    Bucket='compliance-bucket',
    ObjectLockEnabledForBucket=True,
)

# 配置默认保留策略
s3.put_object_lock_configuration(
    Bucket='compliance-bucket',
    ObjectLockConfiguration={
        'ObjectLockEnabled': 'Enabled',
        'Rule': {
            'DefaultRetention': {
                'Mode': 'COMPLIANCE',  # 或 'GOVERNANCE'
                'Days': 365,           # 保留 365 天
            }
        }
    },
)

# 上传对象并设置保留期
retain_until = datetime.now(timezone.utc) + timedelta(days=2555)  # 7 年
s3.put_object(
    Bucket='compliance-bucket',
    Key='audit/2025/financial-report.pdf',
    Body=open('/data/report.pdf', 'rb'),
    ObjectLockMode='COMPLIANCE',
    ObjectLockRetainUntilDate=retain_until,
)

# COMPLIANCE 模式 vs GOVERNANCE 模式：
#
# COMPLIANCE（合规模式）：
# - 任何人（包括根账户）都不能在保留期内删除或覆盖对象
# - 保留期不能缩短，只能延长
# - 适用于法规合规要求（如 SEC 17a-4、HIPAA）
#
# GOVERNANCE（治理模式）：
# - 具有特定 IAM 权限的用户可以绕过保留策略
# - 保留期可以缩短或移除
# - 适用于内部数据保护策略

# 设置法律保留（Legal Hold）
s3.put_object_legal_hold(
    Bucket='compliance-bucket',
    Key='audit/2025/financial-report.pdf',
    LegalHold={'Status': 'ON'},
)
# 法律保留独立于保留期，可以随时添加/移除
# 在法律保留生效期间，对象不能被删除（即使保留期已过）

import boto3

s3 = boto3.client('s3', region_name='us-east-1')

lifecycle_config = {
    'Rules': [
        {
            # 规则 1：日志文件的自动分层与过期
            'ID': 'log-lifecycle',
            'Status': 'Enabled',
            'Filter': {'Prefix': 'logs/'},
            'Transitions': [
                {
                    # 30 天后转为低频访问
                    'Days': 30,
                    'StorageClass': 'STANDARD_IA',
                },
                {
                    # 90 天后转为 Glacier Instant Retrieval
                    'Days': 90,
                    'StorageClass': 'GLACIER_IR',
                },
                {
                    # 365 天后转为 Glacier Deep Archive
                    'Days': 365,
                    'StorageClass': 'DEEP_ARCHIVE',
                },
            ],
            'Expiration': {
                # 2555 天（约 7 年）后自动删除
                'Days': 2555,
            },
        },
        {
            # 规则 2：自动清理未完成的分片上传
            'ID': 'abort-incomplete-uploads',
            'Status': 'Enabled',
            'Filter': {'Prefix': ''},
            'AbortIncompleteMultipartUpload': {
                'DaysAfterInitiation': 7,
            },
        },
        {
            # 规则 3：自动清理过期的删除标记
            'ID': 'cleanup-delete-markers',
            'Status': 'Enabled',
            'Filter': {'Prefix': ''},
            'Expiration': {
                'ExpiredObjectDeleteMarker': True,
            },
        },
        {
            # 规则 4：旧版本的生命周期管理
            'ID': 'version-lifecycle',
            'Status': 'Enabled',
            'Filter': {'Prefix': 'data/'},
            'NoncurrentVersionTransitions': [
                {
                    'NoncurrentDays': 30,
                    'StorageClass': 'STANDARD_IA',
                },
                {
                    'NoncurrentDays': 90,
                    'StorageClass': 'GLACIER',
                },
            ],
            'NoncurrentVersionExpiration': {
                'NoncurrentDays': 365,
                # 保留最近 3 个非当前版本
                'NewerNoncurrentVersions': 3,
            },
        },
        {
            # 规则 5：基于标签过滤
            'ID': 'temp-data-cleanup',
            'Status': 'Enabled',
            'Filter': {
                'Tag': {
                    'Key': 'lifecycle',
                    'Value': 'temporary',
                },
            },
            'Expiration': {
                'Days': 1,
            },
        },
    ],
}

s3.put_bucket_lifecycle_configuration(
    Bucket='my-bucket',
    LifecycleConfiguration=lifecycle_config,
)

import boto3
import json

s3 = boto3.client('s3', region_name='us-east-1')

# 前提条件：
# 1. 源和目标 Bucket 都必须启用版本控制
# 2. 需要创建 IAM 角色授权 S3 执行复制

# 启用源 Bucket 的版本控制
s3.put_bucket_versioning(
    Bucket='source-bucket',
    VersioningConfiguration={'Status': 'Enabled'},
)

# 配置复制规则
replication_config = {
    'Role': 'arn:aws:iam::123456789012:role/s3-replication-role',
    'Rules': [
        {
            'ID': 'replicate-all',
            'Status': 'Enabled',
            'Priority': 1,
            'Filter': {
                'Prefix': '',  # 复制所有对象
            },
            'Destination': {
                'Bucket': 'arn:aws:s3:::destination-bucket',
                'StorageClass': 'STANDARD',
                # 可选：指定不同的存储类别
                # 'StorageClass': 'STANDARD_IA',

                # 可选：跨账户复制需要指定账户 ID
                # 'Account': '987654321098',
                # 'AccessControlTranslation': {
                #     'Owner': 'Destination',
                # },

                # 可选：启用复制时间控制（Replication Time Control）
                'ReplicationTime': {
                    'Status': 'Enabled',
                    'Time': {
                        'Minutes': 15,  # 99.99% 的对象在 15 分钟内复制完成
                    },
                },
                # 可选：启用复制指标
                'Metrics': {
                    'Status': 'Enabled',
                    'EventThreshold': {
                        'Minutes': 15,
                    },
                },
            },
            'DeleteMarkerReplication': {
                'Status': 'Enabled',  # 复制删除标记
            },
        },
        {
            'ID': 'replicate-logs-to-archive',
            'Status': 'Enabled',
            'Priority': 2,
            'Filter': {
                'Prefix': 'logs/',
            },
            'Destination': {
                'Bucket': 'arn:aws:s3:::archive-bucket',
                'StorageClass': 'GLACIER',
            },
            'DeleteMarkerReplication': {
                'Status': 'Disabled',
            },
        },
    ],
}

s3.put_bucket_replication(
    Bucket='source-bucket',
    ReplicationConfiguration=replication_config,
)

import boto3
from datetime import datetime, timezone

cloudwatch = boto3.client('cloudwatch', region_name='us-east-1')


def get_replication_metrics(bucket_name, rule_id, hours=24):
    """查询复制延迟指标"""
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(hours=hours)

    metrics = {
        # 等待复制的操作数
        'OperationsPendingReplication': 'Count',
        # 等待复制的数据量（字节）
        'BytesPendingReplication': 'Bytes',
        # 复制延迟（秒）
        'ReplicationLatency': 'Seconds',
        # 失败的复制操作数
        'OperationsFailedReplication': 'Count',
    }

    for metric_name, unit in metrics.items():
        response = cloudwatch.get_metric_statistics(
            Namespace='AWS/S3',
            MetricName=metric_name,
            Dimensions=[
                {'Name': 'SourceBucket', 'Value': bucket_name},
                {'Name': 'DestinationBucket', 'Value': 'destination-bucket'},
                {'Name': 'RuleId', 'Value': rule_id},
            ],
            StartTime=start_time,
            EndTime=end_time,
            Period=3600,
            Statistics=['Average', 'Maximum'],
        )

        if response['Datapoints']:
            latest = max(response['Datapoints'], key=lambda x: x['Timestamp'])
            print(f"{metric_name}: 平均={latest['Average']:.2f}, 最大={latest['Maximum']:.2f} {unit}")

import boto3
import json

s3 = boto3.client('s3', region_name='us-east-1')

notification_config = {
    # Lambda 函数触发
    'LambdaFunctionConfigurations': [
        {
            'Id': 'process-new-images',
            'LambdaFunctionArn': (
                'arn:aws:lambda:us-east-1:123456789012'
                ':function:image-processor'
            ),
            'Events': ['s3:ObjectCreated:*'],
            'Filter': {
                'Key': {
                    'FilterRules': [
                        {'Name': 'prefix', 'Value': 'uploads/images/'},
                        {'Name': 'suffix', 'Value': '.jpg'},
                    ]
                }
            },
        },
    ],
    # SQS 队列
    'QueueConfigurations': [
        {
            'Id': 'log-processing-queue',
            'QueueArn': (
                'arn:aws:sqs:us-east-1:123456789012:log-processing'
            ),
            'Events': ['s3:ObjectCreated:*'],
            'Filter': {
                'Key': {
                    'FilterRules': [
                        {'Name': 'prefix', 'Value': 'logs/'},
                        {'Name': 'suffix', 'Value': '.gz'},
                    ]
                }
            },
        },
    ],
    # SNS 主题
    'TopicConfigurations': [
        {
            'Id': 'deletion-alerts',
            'TopicArn': (
                'arn:aws:sns:us-east-1:123456789012:s3-deletion-alerts'
            ),
            'Events': ['s3:ObjectRemoved:*'],
        },
    ],
}

s3.put_bucket_notification_configuration(
    Bucket='my-bucket',
    NotificationConfiguration=notification_config,
)

import json
import urllib.parse
import boto3

s3_client = boto3.client('s3')


def lambda_handler(event, context):
    """处理 S3 事件的 Lambda 函数"""
    for record in event['Records']:
        # 解析事件信息
        event_name = record['eventName']
        bucket = record['s3']['bucket']['name']
        key = urllib.parse.unquote_plus(
            record['s3']['object']['key'], encoding='utf-8'
        )
        size = record['s3']['object'].get('size', 0)
        etag = record['s3']['object'].get('eTag', '')

        print(f"事件: {event_name}, 桶: {bucket}, 键: {key}, 大小: {size}")

        # 根据事件类型分发处理
        if event_name.startswith('ObjectCreated'):
            handle_object_created(bucket, key, size)
        elif event_name.startswith('ObjectRemoved'):
            handle_object_removed(bucket, key)


def handle_object_created(bucket, key, size):
    """处理新对象创建"""
    if key.endswith(('.jpg', '.jpeg', '.png')):
        # 生成缩略图
        generate_thumbnail(bucket, key)
    elif key.endswith('.csv'):
        # 触发 ETL 流水线
        trigger_etl_pipeline(bucket, key)
    elif key.endswith('.log.gz'):
        # 索引日志数据
        index_log_data(bucket, key)


def handle_object_removed(bucket, key):
    """处理对象删除"""
    # 清理关联资源（如缩略图、索引等）
    cleanup_related_resources(bucket, key)

import boto3

s3 = boto3.client('s3', region_name='us-east-1')

# 启用 EventBridge 通知
s3.put_bucket_notification_configuration(
    Bucket='my-bucket',
    NotificationConfiguration={
        'EventBridgeConfiguration': {},  # 启用即可，无需额外配置
    },
)

# EventBridge 的优势：
# 1. 支持更丰富的事件过滤（按对象大小、元数据等）
# 2. 支持更多目标（Step Functions、API Gateway、Kinesis 等）
# 3. 支持事件归档和重放
# 4. 支持跨账户事件路由
# 5. 与 CloudTrail 集成，可以基于 API 调用触发

import hashlib


def generate_distributed_key(logical_path, num_prefixes=16):
    """生成分散的对象键"""
    hash_value = hashlib.md5(logical_path.encode()).hexdigest()
    prefix = hash_value[:4]  # 取前 4 个字符作为分散前缀
    return f"{prefix}/{logical_path}"


# 示例
keys = [
    generate_distributed_key(f"events/2025/09/25/event-{i:06d}.json")
    for i in range(100)
]

# 结果示例：
# "a3f2/events/2025/09/25/event-000000.json"
# "7b1c/events/2025/09/25/event-000001.json"
# "d4e8/events/2025/09/25/event-000002.json"
# 请求均匀分布在 65536 个前缀中

import os
import boto3
from boto3.s3.transfer import TransferConfig

s3 = boto3.client('s3', region_name='us-east-1')

# boto3 的 TransferConfig 控制并发行为
transfer_config = TransferConfig(
    multipart_threshold=8 * 1024 * 1024,  # 8MB 以上使用分片上传
    max_concurrency=10,                    # 最大并发数
    multipart_chunksize=8 * 1024 * 1024,  # 每个分片 8MB
    num_download_threads=10,               # 下载线程数
    max_io_queue=100,                      # IO 队列深度
    io_chunksize=256 * 1024,              # IO 块大小
)

# 上传
s3.upload_file(
    '/data/large-file.parquet',
    'my-bucket',
    'data/large-file.parquet',
    Config=transfer_config,
)

# 下载
s3.download_file(
    'my-bucket',
    'data/large-file.parquet',
    '/data/downloaded.parquet',
    Config=transfer_config,
)

# 使用 Range GET 并发下载大文件
from concurrent.futures import ThreadPoolExecutor


def parallel_download(bucket, key, output_path, part_size=64*1024*1024, workers=16):
    """并发 Range GET 下载"""
    head = s3.head_object(Bucket=bucket, Key=key)
    file_size = head['ContentLength']

    # 预分配文件
    with open(output_path, 'wb') as f:
        f.seek(file_size - 1)
        f.write(b'\0')

    def download_range(start, end, part_num):
        response = s3.get_object(
            Bucket=bucket, Key=key,
            Range=f'bytes={start}-{end}'
        )
        data = response['Body'].read()
        with open(output_path, 'r+b') as f:
            f.seek(start)
            f.write(data)
        return part_num, len(data)

    tasks = []
    offset = 0
    part_num = 0
    while offset < file_size:
        end = min(offset + part_size - 1, file_size - 1)
        tasks.append((offset, end, part_num))
        offset = end + 1
        part_num += 1

    with ThreadPoolExecutor(max_workers=workers) as executor:
        futures = [
            executor.submit(download_range, start, end, pn)
            for start, end, pn in tasks
        ]
        for f in futures:
            pn, sz = f.result()
            print(f"分片 {pn} 下载完成: {sz} 字节")

    print(f"下载完成: {output_path} ({file_size / (1024**3):.2f} GB)")

import boto3

s3 = boto3.client('s3', region_name='us-east-1')

# 启用 Transfer Acceleration
s3.put_bucket_accelerate_configuration(
    Bucket='my-bucket',
    AccelerateConfiguration={'Status': 'Enabled'},
)

# 使用加速端点上传
s3_accel = boto3.client(
    's3',
    region_name='us-east-1',
    config=boto3.session.Config(
        s3={'use_accelerate_endpoint': True}
    ),
)

# 上传——SDK 自动使用加速端点
s3_accel.upload_file(
    '/data/large-file.tar.gz',
    'my-bucket',
    'uploads/large-file.tar.gz',
)

# 测试加速效果（AWS 提供的速度比较工具）：
# https://s3-accelerate-speedtest.s3-accelerate.amazonaws.com/en/accelerate-speed-comparsion.html