ShowDocs
搜索文章Ctrl K

外观

← 返回Amazon S3列表
应用实践

S3 实践：带条件写与校验的完整 Boto3 CLI ​

配置和依赖 ​

S3 配置要求读者预先准备 AWS 普通用途私有实验桶、受限角色或 profile、正确 Region 和独立前缀；前置是 IAM及条件写。Python 3.10+、Boto3 1.x 声明如下；示例未执行，仅静态审阅，本轮没有安装、访问账号或启动任何服务。版本范围不是已经验证的精确锁定版本。

txt
# IfMatch/IfNoneMatch refer to current general purpose bucket APIs; use this family.
boto3>=1.40,<2
ini
# Documentation only; Boto3 reads the environment or configured role/profile chain.
AWS_DEFAULT_REGION=us-east-1
S3_BUCKET=replace-with-existing-private-lab-bucket
S3_PREFIX=theory-lab/your-session/
# Optional named lab profile. Do not combine accidentally with stale environment keys.
AWS_PROFILE=
AWS_ACCESS_KEY_ID=
AWS_SECRET_ACCESS_KEY=
AWS_SESSION_TOKEN=

.env.example 只展示变量，不被 CLI 自动加载。使用 profile 时不要同时留着意外优先的环境 AK/SK；使用临时环境凭据时三元组要完整。若使用运行环境角色，可不设静态密钥。S3_PREFIX 仅防误操作，IAM 仍要限制桶与前缀。没有开启版本控制时 version 输出可能为空，本 CLI 不允许绕过版本参数删除。

全部源码 ​

py
"""AWS S3 general purpose bucket sample, not an S3-compatible abstraction."""
import argparse
import base64
import hashlib
import json
import os
import sys
import uuid
from pathlib import Path

import boto3
from botocore.config import Config
from botocore.exceptions import ClientError


def main():
    parser = argparse.ArgumentParser(description="Existing private S3 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()
    client = None
    try:
        prefix = os.environ["S3_PREFIX"]
        if not prefix.startswith("theory-lab/") or not prefix.endswith("/"):
            raise ValueError("S3_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")
        client = boto3.client("s3", region_name=os.environ["AWS_DEFAULT_REGION"],
                              config=Config(signature_version="s3v4", connect_timeout=10,
                                            read_timeout=30, retries={"mode": "standard", "total_max_attempts": 3}))
        bucket = os.environ["S3_BUCKET"]
        params = {"Bucket": bucket, "Key": key}
        if args.version:
            params["VersionId"] = args.version
        if args.command == "list":
            for page in client.get_paginator("list_objects_v2").paginate(
                    Bucket=bucket, Prefix=prefix, PaginationConfig={"PageSize": 100}):
                for item in page.get("Contents", []):
                    print(json.dumps({"key": item["Key"], "size": item["Size"]}))
        elif args.command == "put":
            if Path(args.file).stat().st_size > 1024 * 1024:
                raise ValueError("sample limits uploads to 1 MiB")
            data = Path(args.file).read_bytes()
            checksum = base64.b64encode(hashlib.sha256(data).digest()).decode("ascii")
            result = client.put_object(Bucket=bucket, Key=key, Body=data, IfNoneMatch="*",
                                       ChecksumAlgorithm="SHA256", ChecksumSHA256=checksum)
            print(json.dumps({"key": key, "etag": result["ETag"], "version": result.get("VersionId"),
                              "checksum_sha256": result.get("ChecksumSHA256")}))
        elif args.command == "get":
            if not args.file:
                raise ValueError("get requires --file (must not exist)")
            result = client.get_object(**params, ChecksumMode="ENABLED")
            stream = result["Body"]
            try:
                with open(args.file, "xb") as output:
                    for chunk in stream.iter_chunks(chunk_size=65536):
                        output.write(chunk)
            finally:
                stream.close()
        elif args.command == "sign":
            print(client.generate_presigned_url("get_object", Params=params, ExpiresIn=300))
        else:
            result = client.delete_object(**params)
            print(json.dumps({"version": result.get("VersionId"), "delete_marker": result.get("DeleteMarker")}))
        return 0
    except ClientError as error:
        print(json.dumps({"code": error.response["Error"]["Code"],
                          "request_id": error.response["ResponseMetadata"].get("RequestId")}), file=sys.stderr)
        return 1
    except Exception as error:
        print(type(error).__name__ + ": operation failed; check inputs/credential chain", file=sys.stderr)
        return 1
    finally:
        if client is not None:
            client.close()


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

S3 完整源码通过 paginator 遍历所有 list_objects_v2 页，上传发送 IfNoneMatch='*' 和显式 SHA-256；GET 读取 StreamingBody 并在 finally 关闭，最终 client.close 释放连接池。本地文件使用上下文管理且拒绝覆盖。成功退出 0，argparse 错误退出 2，配置、ClientError、传输和文件错误退出 1。远端失败只打印代码和 RequestId，不打印响应全文或密钥。

程序每次 put 生成 UUID，上传限制 1 MiB 并在内存中计算校验和，避免把大文件吞进内存。get 未设置下载大小上限，且失败可能留下部分文件；业务应采用临时文件、摘要检查与发布步骤。get 不传 version 时读取当前对象，需要重现上传内容就应传 VersionId。源码中的 AWS 客户端不接受其他产品端点。Boto3 PutObject 参数与Delete 合同是核对来源。

读者自行运行与预期 ​

下列步骤仅供读者在独立受限环境自行执行，本轮没有执行。先人工确认资源、Region、权限和成本，再在仓库根目录创建虚拟环境：

sh
python -m venv examples/s3/.venv
# 激活 .venv；Windows 使用对应 Scripts/Activate.ps1
python -m pip install -r examples/s3/requirements.txt
# 在当前 shell 设置 Region、桶、前缀和获准的凭据来源。
python examples/s3/object_cli.py list
python examples/s3/object_cli.py put --file ./sample.txt
python examples/s3/object_cli.py get --key theory-lab/your-session/RETURNED-UUID.bin --version RETURNED-VERSION --file ./download.bin
python examples/s3/object_cli.py sign --key theory-lab/your-session/RETURNED-UUID.bin --version RETURNED-VERSION

S3 上传预期输出 key、etag、version 和 checksum_sha256；list 返回当前可见对象，空前缀无输出。下载后自行比对原文件与下载文件 SHA-256，签名输出五分钟 URL。与打印 JSON 不同，实际打开该 URL 是另一次产品请求，受读取与 KMS 权限约束并可能计费；不能将链接生成作为成功下载证据。

精确清理与练习 ​

S3 精确版本删除在人工核对后自行运行 python examples/s3/object_cli.py delete --key KEY --version VERSION，会永久删除那个版本，可能受 Object Lock 或策略限制。程序不清空桶、不列举删除标记、不自动清理 multipart；未版本化桶的清理由读者使用受限工具处理。请求、流量和历史存储均可能计费，单文件上限不代表预算上限。

练习：重复 put 为何不产生412？答案：程序每次换 UUID，两个请求不竞争同名目标，条件保护的是单个名称。若业务要求同一请求只产生一个对象，应持久化请求键与既有对象引用。练习：下载返回200后为何还要比较业务摘要？答案：传输完整性不验证业务映射，程序可能请求到另一份合法对象。练习：只捕获 ClientError 足够吗？答案：本地 IO 和网络等错误也需要失败退出并释放资源。

继续阅读