ShowDocs
搜索文章Ctrl K

外观

← 返回阿里云 OSS列表
应用实践

OSS 实践：完整原生 SDK CLI 阅读项目 ​

项目合同与配置 ​

前置是 防覆盖边界 与一个由读者独立准备的私有实验桶。本项目只供阅读和读者自行实验；本轮示例未执行，没有安装依赖、启动服务、访问账号或调用产品 API。Python 3.10+、SDK V1 oss2 2.x 的依赖声明如下；没有为本站增加产品依赖。

txt
# SDK V1 (oss2), not alibabacloud-oss-v2. Range is a declared family, not a lockfile.
oss2>=2.19,<3
ini
# Documentation only; the CLI does not parse or source this file.
OSS_ENDPOINT=https://oss-cn-hangzhou.aliyuncs.com
OSS_REGION=cn-hangzhou
OSS_BUCKET=replace-with-existing-private-lab-bucket
OSS_PREFIX=theory-lab/your-session/
OSS_ACCESS_KEY_ID=
OSS_ACCESS_KEY_SECRET=
OSS_SESSION_TOKEN=

OSS 配置要求 Endpoint 与 region 一致，prefix 必须以 theory-lab/ 开头并以 / 结尾。.env.example 不会被自动加载；读者应在自己的 shell 中逐项导出，密钥来自受限 RAM 或临时 STS 凭据。空 token 用于长期密钥的输入格式，不表示推荐长期密钥。示例不检查桶是否版本化，必须先由读者确认；否则 put 的 version 输出可能为空，便不能使用本 CLI 的版本删除命令。

全部实现 ​

py
"""OSS Python SDK V1 reading sample. Not executed during authoring."""
import argparse
import json
import os
import sys
import uuid
from pathlib import Path

import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider


def main():
    parser = argparse.ArgumentParser(description="Existing private OSS lab bucket only")
    parser.add_argument("command", choices=["list", "put", "get", "sign", "delete"])
    parser.add_argument("--key")
    parser.add_argument("--file")
    parser.add_argument("--version")
    args = parser.parse_args()
    session = None
    try:
        prefix = os.environ["OSS_PREFIX"]
        if not prefix.startswith("theory-lab/") or not prefix.endswith("/"):
            raise ValueError("OSS_PREFIX must start theory-lab/ and end /")
        key = args.key
        if args.command != "list" and args.command != "put" and not key:
            raise ValueError("--key is required")
        if args.command == "put":
            if not args.file:
                raise ValueError("put requires --file")
            key = prefix + str(uuid.uuid4()) + ".bin"
        if key and (not key.startswith(prefix) or ".." in key.split("/")):
            raise ValueError("key outside configured prefix")
        if args.command == "delete" and not args.version:
            raise ValueError("delete requires explicit --version; no current-version delete")
        endpoint = os.environ["OSS_ENDPOINT"]
        if not endpoint.startswith("https://"):
            raise ValueError("HTTPS endpoint required")
        session = oss2.Session()
        bucket = oss2.Bucket(
            oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider()),
            endpoint, os.environ["OSS_BUCKET"], region=os.environ["OSS_REGION"],
            session=session, enable_crc=True, connect_timeout=10,
        )
        params = {"versionId": args.version} if args.version else None
        if args.command == "list":
            for item in oss2.ObjectIteratorV2(bucket, prefix=prefix, max_keys=100):
                print(json.dumps({"key": item.key, "size": item.size}, ensure_ascii=False))
        elif args.command == "put":
            if Path(args.file).stat().st_size > 1024 * 1024:
                raise ValueError("sample limits uploads to 1 MiB")
            with open(args.file, "rb") as stream:
                result = bucket.put_object(key, stream, headers={"x-oss-forbid-overwrite": "true"})
            print(json.dumps({"key": key, "etag": result.etag,
                              "version": result.headers.get("x-oss-version-id"),
                              "request_id": result.request_id}))
        elif args.command == "get":
            if not args.file:
                raise ValueError("get requires --file (must not exist)")
            result = bucket.get_object(key, params=params)
            try:
                with open(args.file, "xb") as output:
                    while True:
                        chunk = result.read(65536)
                        if not chunk:
                            break
                        output.write(chunk)
            finally:
                result.resp.response.close()
        elif args.command == "sign":
            print(bucket.sign_url("GET", key, 300, params=params, slash_safe=True))
        else:
            result = bucket.delete_object(key, params=params)
            print(json.dumps({"status": result.status, "request_id": result.request_id}))
        return 0
    except oss2.exceptions.OssError as error:
        print(json.dumps({"status": error.status, "code": error.code,
                          "request_id": error.request_id}), file=sys.stderr)
        return 1
    except Exception as error:
        print(type(error).__name__ + ": operation failed; check local inputs/credentials", file=sys.stderr)
        return 1
    finally:
        if session is not None:
            session.session.close()


if __name__ == "__main__":
    sys.exit(main())

OSS 完整源码包括 argparse 入口、ObjectIteratorV2 自动分页、文件上下文、响应流关闭、显式会话连接池关闭和失败退出码。V1 没有统一 bucket.close()，这里依据 SDK HTTP 封装源码关闭自己创建的 session.session；下载响应通过 result.resp.response.close() 释放。它依赖 V1 类型布局，迁到 V2 时必须重新核对，不能只换依赖名。

程序成功退出 0，参数解析失败退出 2，配置或远端错误退出 1。错误输出保留服务端 code、status、request_id，不打印密钥或响应全文。下载使用 xb，目标存在即失败，且中途失败可能留下不完整文件；业务应用应先下载到临时文件，核对完整性后原子发布。CLI 的 list 可读取所有匹配页，不能推断它只发送一次请求。

读者自行运行与预期结果 ​

下列命令由读者在自己的受限环境执行，本轮没有执行。先人工确认实验桶、权限、预算和独立前缀，再在仓库根目录使用独立环境：

sh
python -m venv examples/oss/.venv
# 激活该虚拟环境后执行；Windows 使用 .venv/Scripts/Activate.ps1
python -m pip install -r examples/oss/requirements.txt
# 用当前 shell 设置 .env.example 中的变量，不要将密钥写进命令历史。
python examples/oss/object_cli.py list
python examples/oss/object_cli.py put --file ./sample.txt
python examples/oss/object_cli.py get --key theory-lab/your-session/RETURNED-UUID.bin --version RETURNED-VERSION --file ./download.bin
python examples/oss/object_cli.py sign --key theory-lab/your-session/RETURNED-UUID.bin --version RETURNED-VERSION

OSS 上传预期输出随机 key、ETag、request_id，以及版本桶的 version；list 返回当前可见对象，get 生成新本地文件，sign 输出一条五分钟 URL。下载后读者可自行对比原文件和下载文件的 SHA-256。若版本未启用，可省略 get 的 version，但不能假设读到的是上传时那个内容；更不能把省略版本的删除加进脚本。

清理、费用与练习 ​

OSS 精确版本删除会永久移除指定版本，本项目要求显式 --version，只可在人工核对输出后自行运行 python examples/oss/object_cli.py delete --key KEY --version VERSION。它不会清空整个前缀，也不会枚举并删除所有历史版本；未版本桶的清理由读者用受限工具人工完成。下载、列表和保留版本都可能收费，签名生成本身不证明被签名操作会免费或成功。

练习一：改用固定 key 后重复上传是否就是幂等？答案：不是；版本桶可能生成多个版本，非版本桶可能触发 409，网络失败还可能让提交结果不明。练习二：前缀检查为何不构成安全边界？答案：有权限的用户可绕过此程序直接请求 OSS，真正边界在 RAM、桶策略与密钥管理。练习三：签名 URL 成功生成但 GET 404 是否矛盾？答案：本地签名不读取对象，它可以为不存在的 key 生成 URL。

继续阅读