aliyun-oss-x

阿里云 OSS Python SDK（支持异步）

English README

概述

基于 httpx 的阿里云对象存储 Python SDK，支持异步操作并提供类型提示。

[!Note]

• 此版本不包含 osscmd 命令行工具。
• 此版本仅支持 Python 3.10 及以上版本。
• 此版本仅支持 V4 签名。

运行环境

Python 3.10 及以上版本。

安装

通过 PIP 安装官方发布版本：

pip install aliyun-oss-x

如果需要使用阿里云 KMS 加密，请安装 aliyun-kms 扩展包：

pip install aliyun-oss-x[aliyun-kms]

快速入门

同步用法

import aliyun_oss_x

endpoint = 'http://oss-cn-hangzhou.aliyuncs.com' # 假设你的存储桶位于杭州地区

auth = aliyun_oss_x.Auth('<你的 AccessKeyID>', '<你的 AccessKeySecret>')
bucket = aliyun_oss_x.Bucket(auth, endpoint, '<你的存储桶名称>')

# 存储桶中的对象键为 story.txt
key = 'story.txt'

# 上传
bucket.put_object(key, 'Ali Baba 是一个快乐的青年。')

# 下载
bucket.get_object(key).read()

# 删除
bucket.delete_object(key)

# 遍历存储桶中的所有对象
for object_info in aliyun_oss_x.ObjectIterator(bucket):
    print(object_info.key)

异步用法

import asyncio

import aliyun_oss_x

endpoint = 'http://oss-cn-hangzhou.aliyuncs.com' # 假设你的存储桶位于杭州地区

auth = aliyun_oss_x.Auth('<你的 AccessKeyID>', '<你的 AccessKeySecret>')
bucket = aliyun_oss_x.AsyncBucket(auth, endpoint, '<你的存储桶名称>', region="cn-hangzhou")

async def main():
    # 存储桶中的对象键为 story.txt
    key = 'story.txt'

    # 上传
    await bucket.put_object(key, 'Ali Baba 是一个快乐的青年。')

    # 下载
    await bucket.get_object(key).read()

    # 删除
    await bucket.delete_object(key)

    # 遍历存储桶中的所有对象
    async for object_info in aliyun_oss_x.AsyncObjectIterator(bucket):
        print(object_info.key)

asyncio.run(main())

更多示例请参考 "examples" 目录下的代码。

错误处理

除非另有说明，Python SDK 接口在出错时会抛出异常（参见 aliyun_oss_x.exceptions 子模块）。以下是一个示例：

try:
    result = bucket.get_object(key)
    print(result.read())
except aliyun_oss_x.exceptions.NoSuchKey as e:
    print('{0} 未找到：http_status={1}, request_id={2}'.format(key, e.status, e.request_id))

设置日志

以下代码可以设置 'aliyun_oss_x' 的日志级别：

import logging
logging.getLogger('aliyun_oss_x').setLevel(logging.WARNING)

测试

首先通过环境变量设置测试所需的 AccessKeyId、AccessKeySecret、endpoint 和 bucket 信息（不要使用生产环境的存储桶）。 以 Linux 系统为例：

export OSS_TEST_ACCESS_KEY_ID=<AccessKeyId>
export OSS_TEST_ACCESS_KEY_SECRET=<AccessKeySecret>
export OSS_TEST_ENDPOINT=<endpoint>
export OSS_TEST_BUCKET=<bucket>

export OSS_TEST_STS_ID=<用于测试 STS 的 AccessKeyId>
export OSS_TEST_STS_KEY=<用于测试 STS 的 AccessKeySecret>
export OSS_TEST_STS_ARN=<用于测试 STS 的角色 ARN>

按以下方式运行测试：

nosetests                          # 首先安装 nose

更多资源

• 更多示例
• Python SDK API 文档
• 官方 Python SDK 文档

联系我们

• 阿里云 OSS 官方网站
• 阿里云 OSS 官方论坛
• 阿里云 OSS 官方文档中心
• 阿里云官方技术支持：提交工单

许可证

• MIT