aod-oss-client

阿里云 OSS 上传客户端，同时支持 同步 和 异步 两种接口。

特性

• 同步 + 异步双 API，同一 OssClient 实例可多次复用
• 自动选择简单上传（< 100MB）或分片上传（≥ 100MB）
• 分片并发上传，内置指数退避重试
• content_type 自动检测，不传自动推断
• skip_if_exists 支持幂等跳过
• 凭证支持显式传入 / 环境变量 / .env 文件
• 支持自定义域名、内网 Endpoint

安装

pip install aod-oss-client
# 或
uv add aod-oss-client

快速开始

同步用法

from aod_oss_client import OssClient, OssClientConfig

config = OssClientConfig(bucket="my-bucket", region="cn-hangzhou")
client = OssClient(config)

r1 = client.upload_file("/data/a.geojson", "data/a.geojson")
r2 = client.upload_file("/data/b.fgb", "data/b.fgb")

print(r1.url, r2.url)
# client.close()  # 可选

异步用法

from aod_oss_client import OssClient, OssClientConfig

config = OssClientConfig(bucket="my-bucket", region="cn-hangzhou")
client = OssClient(config)

r1 = await client.async_upload_file("/data/a.geojson", "data/a.geojson")
r2 = await client.async_upload_file("/data/b.fgb", "data/b.fgb")

print(r1.url, r2.url)
# await client.async_aclose()  # 可选

幂等跳过

result = client.upload_file("photo.jpg", "images/photo.jpg", skip_if_exists=True)
if result.skipped:
    print("文件已存在，跳过上传")

自定义元属性

上传时可通过 metadata 参数设置 OSS 对象自定义元属性（即 x-oss-meta-* 头）：

client.upload_file(
    "data.geojson", "data.geojson",
    metadata={
        "source": "gps_collector",
        "crs": "EPSG:4326",
        "version": "2.1",
    },
)

元属性会在 OSS 控制台和 GetObject 响应的 x-oss-meta-* 头中返回。

content_type 自动检测

不传则根据文件扩展名自动推断：

client.upload_file("data.json", "data.json")                 # → application/json
client.upload_file("data.geojson", "data.geojson")           # → application/geo+json
client.upload_file("data.bin", "data.bin")                   # → application/octet-stream
client.upload_file("data.fgb", "data.fgb", content_type="application/geo+json")  # 手动指定

with 语句

with OssClient(config) as client:
    client.upload_file("a.txt", "a.txt")
    client.upload_file("b.txt", "b.txt")

配置

OssClientConfig

config = OssClientConfig(
    bucket="my-bucket",
    region="cn-hangzhou",
    access_key_id="your-key-id",       # 可选，不传走环境变量
    access_key_secret="your-key-secret",
    endpoint=None,                      # 自定义 Endpoint
    public_base_url=None,               # 自定义域名
    use_internal_endpoint=False,         # 内网 Endpoint
    multipart_threshold=100 * 1024 * 1024,  # 100MB 分片阈值
    part_size=10 * 1024 * 1024,             # 10MB 分片大小
    part_parallel=3,                        # 分片并发数
)

环境变量

OSS_ACCESS_KEY_ID=your_access_key_id
OSS_ACCESS_KEY_SECRET=your_access_key_secret
OSS_REGION=cn-hangzhou
OSS_BUCKET=my-bucket
OSS_ENDPOINT=
OSS_PUBLIC_BASE_URL=
OSS_USE_INTERNAL_ENDPOINT=false

API 参考

OssClient

UploadResult

开发

# 同步依赖
uv sync --all-extras

# 运行测试（单元测试）
uv run pytest

# 真实上传测试（需配置环境变量）
uv run python tests/test_real_upload.py

# 代码检查
uv run ruff check src
uv run ruff format src

License

MIT