ShowDocs
搜索文章Ctrl K

外观

← 返回华为云 OBS列表
应用实践

OBS 实践：完整原生 Python SDK CLI ​

配置和前置 ​

OBS 配置需要读者预先准备的私有普通对象桶、原生 Endpoint、受限 AK/SK 或临时三元组，以及独立实验前缀。前置：原生鉴权和并发边界。Python 3.10+，分发包 esdk-obs-python 固定3.24.3；本轮仅核对合同，示例未执行、未安装、未连接账号。

txt
# Internal TLS context layout was inspected for this exact SDK source version.
# Distribution name differs from its Python import (obs).
esdk-obs-python==3.24.3
ini
# Documentation only; not automatically loaded.
OBS_ENDPOINT=https://obs.cn-north-4.myhuaweicloud.com
OBS_BUCKET=replace-with-existing-private-lab-bucket
OBS_PREFIX=theory-lab/your-session/
OBS_ACCESS_KEY_ID=
OBS_SECRET_ACCESS_KEY=
OBS_SECURITY_TOKEN=
# Optional trusted PEM CA bundle path. Empty/unset uses Python system trust.
OBS_CA_BUNDLE=

此环境文件是说明书，不会自动加载。OBS_ENDPOINT 是服务地址，OBS_SECURITY_TOKEN 只在使用同次颁发的临时 AK/SK 时填写；不要把别家 token 字段直接替换进来。参数只保护实验前缀，真正权限必须由 IAM 与桶策略限制。若桶未开启多版本，上传 version 可能为空，本程序不会提供不带版本的删除退路。

全部源码与错误合同 ​

py
"""Native OBS SDK 3.x, explicit response status handling; not Boto3."""
import argparse
import json
import os
import ssl
import sys
import uuid
from pathlib import Path

from obs import ObsClient


def checked(response):
    if response.status >= 300:
        raise RuntimeError(json.dumps({"status": response.status, "code": response.errorCode,
                                       "request_id": response.requestId}))
    return response


def main():
    parser = argparse.ArgumentParser(description="Existing private OBS object bucket only")
    parser.add_argument("command", choices=["list", "put", "get", "delete", "sign"])
    parser.add_argument("--key")
    parser.add_argument("--file")
    parser.add_argument("--version")
    args = parser.parse_args()
    client = None
    try:
        prefix = os.environ["OBS_PREFIX"]
        if not prefix.startswith("theory-lab/") or not prefix.endswith("/"):
            raise ValueError("OBS_PREFIX must start theory-lab/ and end /")
        key = args.key
        if args.command == "put":
            if not args.file:
                raise ValueError("put requires --file")
            key = prefix + str(uuid.uuid4()) + ".bin"
        elif args.command != "list" and not key:
            raise ValueError("--key is required")
        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")
        endpoint = os.environ["OBS_ENDPOINT"]
        if not endpoint.startswith("https://"):
            raise ValueError("HTTPS endpoint required")
        tls_context = ssl.create_default_context(cafile=os.environ.get("OBS_CA_BUNDLE") or None)
        client = ObsClient(access_key_id=os.environ["OBS_ACCESS_KEY_ID"],
                           secret_access_key=os.environ["OBS_SECRET_ACCESS_KEY"],
                           security_token=os.environ.get("OBS_SECURITY_TOKEN") or None,
                           server=endpoint, signature="obs", timeout=30,
                           ssl_verify=True, use_http2=False, long_conn_mode=False,
                           is_signature_negotiation=False)
        # Pinned 3.24.3 HTTP/1.x consumes this context on the FIRST request.
        # Its built-in ssl_verify path does not enable hostname checking.
        # Constructor performs no network request with long_conn_mode=False.
        # Internal-field dependency: re-audit before changing the SDK version.
        client.context = tls_context
        bucket = os.environ["OBS_BUCKET"]
        if args.command == "list":
            marker = None
            while True:
                response = checked(client.listObjects(bucket, prefix=prefix, marker=marker, max_keys=100))
                for item in response.body.contents:
                    print(json.dumps({"key": item.key, "size": item.size}))
                if not response.body.is_truncated:
                    break
                next_marker = response.body.next_marker
                if not next_marker or next_marker == marker:
                    raise RuntimeError("pagination did not advance")
                marker = next_marker
        elif args.command == "put":
            if Path(args.file).stat().st_size > 1024 * 1024:
                raise ValueError("sample limits uploads to 1 MiB")
            response = checked(client.putContent(bucket, key, Path(args.file).read_bytes()))
            print(json.dumps({"key": key, "etag": response.body.etag,
                              "version": response.body.versionId, "request_id": response.requestId}))
        elif args.command == "get":
            if not args.file:
                raise ValueError("get requires --file (must not exist)")
            response = checked(client.getObject(bucket, key, versionId=args.version))
            stream = response.body.response
            try:
                with open(args.file, "xb") as output:
                    while True:
                        chunk = stream.read(65536)
                        if not chunk:
                            break
                        output.write(chunk)
            finally:
                stream.close()
        elif args.command == "sign":
            query = {"versionId": args.version} if args.version else None
            print(client.createSignedUrl("GET", bucket, key, expires=300, queryParams=query).signedUrl)
        else:
            response = checked(client.deleteObject(bucket, key, versionId=args.version))
            print(json.dumps({"status": response.status, "request_id": response.requestId}))
        return 0
    except RuntimeError as error:
        print(str(error), file=sys.stderr)
        return 1
    except Exception as error:
        print(type(error).__name__ + ": check inputs, endpoint and native OBS credentials", file=sys.stderr)
        return 1
    finally:
        if client is not None:
            client.close()


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

OBS 完整源码在每个远端响应上调用 checked，先看 status，再读取 body；listObjects 通过 is_truncated 与 next_marker 翻页，并防止 marker 不前进导致死循环。下载读取 response 流且在 finally 关闭，ObsClient 最终 close。成功退出 0，命令格式错误退出 2，配置、鉴权、远端非 2xx 和本地 IO 错误退出 1。只输出诊断代码与 requestId，不主动打印密钥或 HTTP 请求全文。

OBS TLS 不能仅凭 URL 以 HTTPS 开头就认为证书和主机名都受检。此 SDK 3.24.3 默认 ssl_verify=False，且其内置上下文路径不启用 hostname 检查；示例显式关闭 HTTP/2 和长连接、关闭签名协商，在任何产品调用前将 client.context 替换为 ssl.create_default_context()，同时启用可信 CA 校验与主机名检查。可选 OBS_CA_BUNDLE 指定受信 PEM CA 文件；未设置使用 Python 系统信任。依据是固定版本官方 client.py：构造阶段不发请求，HTTP/1.x 建连把该 context 交给 HTTPSConnection。这是对 SDK 内部字段的明确依赖，升级必须重新审阅；本轮没有执行 TLS 握手验证。

上传每次生成随机 key，读入最大 1 MiB 文件；不是分段上传器。下载用 xb 拒绝覆盖本地文件，中断可能留下部分内容；后续应由读者删除失败文件或实现临时文件发布流程。get 可按版本读取，sign 的 versionId 通过 queryParams 进入签名；没有版本时针对当前内容。源码核对来源：原生对象操作样例及marker 分页文档。

读者自跑与预期结果 ​

下面是读者自行选择的步骤，本轮完全没有执行。先确认桶类型、权限、预算和独立前缀，再在仓库根目录使用虚拟环境：

sh
python -m venv examples/obs/.venv
# 激活 .venv；Windows 对应 Scripts/Activate.ps1
python -m pip install -r examples/obs/requirements.txt
# 在当前 shell 设置环境文件中的变量，不自动登录或修改权限。
python examples/obs/object_cli.py list
python examples/obs/object_cli.py put --file ./sample.txt
python examples/obs/object_cli.py get --key theory-lab/your-session/RETURNED-UUID.bin --version RETURNED-VERSION --file ./download.bin
python examples/obs/object_cli.py sign --key theory-lab/your-session/RETURNED-UUID.bin --version RETURNED-VERSION

OBS 上传预期返回 key、etag、version 与 request_id；list 逐行输出前缀内当前可见对象，空前缀无输出。下载后应单独比较源文件与目标文件 SHA-256，不能仅看退出 0 就推导业务摘要一致。签名预期返回五分钟原生 GET URL，使用它会发生真实读取并可能计费；生成链接本身不证明远端请求会成功。

清理和练习 ​

OBS 精确版本删除只能在人工确认后自行执行 python examples/obs/object_cli.py delete --key KEY --version VERSION，它永久删除选定版本，不清空其他历史版本、标记或分段。请求、流量、历史容量与归档取回可能计费；小文件限制不等于整个实验账单上限。没有版本 ID 的对象由读者用受限工具人工清理，本项目不创建清理资源或自动部署脚本。

练习：为什么 checked 不应该直接包住 createSignedUrl？答案是后者返回本地签名结果，而非带远端 status 的服务响应。为什么 403 后不能直接读取 body.contents？答案是失败响应的数据结构不同，忽略 status 会把权限问题变成属性异常，丢失真正的 requestId。为什么 UUID 不等于幂等？答案是重试生成另一个名字，需要业务请求记录关联既有上传。

继续阅读