如何使用Python的aioboto3库连接并配置Yandex S3存储

前置准备 #

• 安装依赖：执行pip install aioboto3 botocore，aioboto3依赖botocore处理S3协议交互
• 获取Yandex S3凭证：在Yandex云控制台生成访问密钥ID和秘密访问密钥；同时确认存储区域（如ru-central1），对应端点格式为https://<区域>.storage.yandexcloud.net，通用端点为https://storage.yandexcloud.net

客户端初始化 #

aioboto3是异步版本的boto3，需通过异步会话创建客户端：

import aioboto3

async def get_s3_client():
    session = aioboto3.Session()
    async with session.client(
        service_name='s3',
        endpoint_url='https://ru-central1.storage.yandexcloud.net',  # 替换为你的区域端点
        aws_access_key_id='你的访问密钥ID',
        aws_secret_access_key='你的秘密访问密钥',
        region_name='ru-central1'  # 替换为你的存储区域
    ) as client:
        return client

常用操作示例 #

1. 创建存储桶 #

Yandex S3存储桶名称需全局唯一，且需指定区域约束：

async def create_bucket(client, bucket_name):
    await client.create_bucket(
        Bucket=bucket_name,
        CreateBucketConfiguration={'LocationConstraint': 'ru-central1'}
    )
    print(f"存储桶 {bucket_name} 创建成功")

2. 上传文件 #

支持本地文件流式上传：

async def upload_file(client, bucket_name, local_path, s3_key):
    with open(local_path, 'rb') as f:
        await client.put_object(
            Bucket=bucket_name,
            Key=s3_key,
            Body=f
        )
    print(f"文件 {local_path} 已上传至 {bucket_name}/{s3_key}")

3. 下载文件 #

异步读取S3对象流并写入本地：

async def download_file(client, bucket_name, s3_key, local_path):
    response = await client.get_object(Bucket=bucket_name, Key=s3_key)
    async with response['Body'] as stream:
        with open(local_path, 'wb') as f:
            while chunk := await stream.read(4096):
                f.write(chunk)
    print(f"文件 {s3_key} 已下载至 {local_path}")

4. 列出存储桶内对象 #

获取存储桶中的所有文件列表：

async def list_objects(client, bucket_name):
    resp = await client.list_objects_v2(Bucket=bucket_name)
    if 'Contents' in resp:
        print(f"存储桶 {bucket_name} 内对象：")
        for obj in resp['Contents']:
            print(f"- {obj['Key']} (大小: {obj['Size']} 字节)")
    else:
        print(f"存储桶 {bucket_name} 为空")

5. 删除对象 #

删除存储桶中指定文件：

async def delete_object(client, bucket_name, s3_key):
    await client.delete_object(Bucket=bucket_name, Key=s3_key)
    print(f"对象 {s3_key} 已从 {bucket_name} 删除")

完整使用示例 #

整合上述操作的异步入口：

import asyncio

async def main():
    client = await get_s3_client()
    
    # 替换为你的存储桶名称
    bucket_name = "my-yandex-test-bucket"
    
    await create_bucket(client, bucket_name)
    await upload_file(client, bucket_name, "local-test.txt", "remote/test.txt")
    await list_objects(client, bucket_name)
    await download_file(client, bucket_name, "remote/test.txt", "downloaded-test.txt")
    await delete_object(client, bucket_name, "remote/test.txt")
    await list_objects(client, bucket_name)

if __name__ == "__main__":
    asyncio.run(main())

关键注意事项 #

• 权限配置：确保你的访问密钥拥有对应操作的权限（如s3:CreateBucket、s3:PutObject），可在Yandex云IAM控制台中为密钥关联合适的角色。
• 异常处理：实际生产环境需捕获botocore.exceptions.ClientError，处理权限不足、存储桶不存在、对象不存在等异常场景。
• 端点匹配：区域端点需与存储区域一致，避免跨区域访问带来的延迟问题。

内容的提问来源于stack exchange，提问作者Ryabchenko Alexander