简单下载（Python SDK V2）

当从Bucket中下载单个文件（Object）时，您可以指定基于文件最后修改时间或ETag（文件内容标识符）的条件限制。只有当这些条件得到满足时才会执行下载操作；如果不满足，则会返回错误并且不会触发下载。利用限定条件下载不仅可以减少不必要的网络传输和资源消耗，还能提高下载效率。

OSS支持的限定条件如下：

以下示例代码展示了如何使用限定条件下载。

import argparse
import alibabacloud_oss_v2 as oss
from datetime import datetime, timezone

# 创建一个命令行参数解析器，并描述脚本用途：获取对象并保存到文件示例
parser = argparse.ArgumentParser(description="get object to file sample")

# 添加命令行参数 --region，表示存储空间所在的区域，必需参数
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# 添加命令行参数 --bucket，表示要获取对象的存储空间名称，必需参数
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# 添加命令行参数 --endpoint，表示其他服务可用来访问OSS的域名，非必需参数
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
# 添加命令行参数 --key，表示对象（文件）在OSS中的键名，必需参数
parser.add_argument('--key', help='The name of the object.', required=True)
# 添加命令行参数 --file_path，表示下载文件的本地路径，必需参数
parser.add_argument('--file_path', help='The path of the file to save the downloaded content.', required=True)

def main():
    # 解析命令行提供的参数，获取用户输入的值
    args = parser.parse_args()

    # 从环境变量中加载访问OSS所需的认证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK的默认配置创建配置对象，并设置认证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    
    # 设置配置对象的区域属性，根据用户提供的命令行参数
    cfg.region = args.region

    # 如果提供了自定义endpoint，则更新配置对象中的endpoint属性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS客户端，准备与OSS交互
    client = oss.Client(cfg)

    # 定义 if_modified_since 时间
    # 只有在此时间之后被修改的对象才会被返回
    if_modified_since = datetime(2024, 10, 1, 12, 0, 0, tzinfo=timezone.utc)

    # 假设ETag为DA5223EFCD7E0353BE08866700000000，则填写的ETag与Object的ETag值相等时，将满足IfMatch的限定条件，并触发下载行为。
    etag = "\"DA5223EFCD7E0353BE08866700000000\""

    # 执行获取对象并保存到本地文件的请求
    result = client.get_object_to_file(
        oss.GetObjectRequest(
            bucket=args.bucket,  # 指定存储空间名称
            key=args.key,        # 指定对象键名
            if_modified_since=if_modified_since,  # 只有在指定时间之后被修改的对象才会被返回
            if_match=etag,       # 只有 ETag 匹配的对象才会被返回
        ),
        args.file_path  # 指定下载文件的本地路径
    )

    # 输出获取对象的结果信息，包括状态码、请求ID等
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content length: {result.content_length},'
          f' content range: {result.content_range},'
          f' content type: {result.content_type},'
          f' etag: {result.etag},'
          f' last modified: {result.last_modified},'
          f' content md5: {result.content_md5},'
          f' cache control: {result.cache_control},'
          f' content disposition: {result.content_disposition},'
          f' content encoding: {result.content_encoding},'
          f' expires: {result.expires},'
          f' hash crc64: {result.hash_crc64},'
          f' storage class: {result.storage_class},'
          f' object type: {result.object_type},'
          f' version id: {result.version_id},'
          f' tagging count: {result.tagging_count},'
          f' server side encryption: {result.server_side_encryption},'
          f' server side data encryption: {result.server_side_data_encryption},'
          f' next append position: {result.next_append_position},'
          f' expiration: {result.expiration},'
          f' restore: {result.restore},'
          f' process status: {result.process_status},'
          f' delete marker: {result.delete_marker},'
          f' server time: {result.headers.get("x-oss-server-time")},'
    )

# 当此脚本被直接执行时，调用main函数开始处理逻辑
if __name__ == "__main__":
    main()  # 脚本入口点，控制程序流程从这里开始

当您在下载文件时，可以使用进度条实时了解下载进度，避免因为等待时间过长而感到不安或怀疑任务是否卡住。

以下示例代码展示了在下载文件到本地时，使用进度条查看下载文件的进度，以get_object_to_file为例。

import argparse
import alibabacloud_oss_v2 as oss

# 创建一个命令行参数解析器，并描述脚本用途：获取对象示例
parser = argparse.ArgumentParser(description="get object sample")

# 添加命令行参数 --region，表示存储空间所在的区域，必需参数
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# 添加命令行参数 --bucket，表示要获取对象的存储空间名称，必需参数
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# 添加命令行参数 --endpoint，表示其他服务可用来访问OSS的域名，非必需参数
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
# 添加命令行参数 --key，表示对象（文件）在OSS中的键名，必需参数
parser.add_argument('--key', help='The name of the object.', required=True)

def main():
    # 解析命令行提供的参数，获取用户输入的值
    args = parser.parse_args()

    # 从环境变量中加载访问OSS所需的认证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK的默认配置创建配置对象，并设置认证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 设置配置对象的区域属性，根据用户提供的命令行参数
    cfg.region = args.region

    # 如果提供了自定义endpoint，则更新配置对象中的endpoint属性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS客户端，准备与OSS交互
    client = oss.Client(cfg)

    # 定义一个字典变量 progress_state 用于保存下载进度状态，初始值为 0
    progress_state = {'saved': 0}
    
    # 定义进度回调函数 _progress_fn
    def _progress_fn(n, written, total):
        # 使用字典存储累计写入的字节数
        progress_state['saved'] += n

        # 计算当前下载百分比，将已写入字节数与总字节数进行除法运算后取整
        rate = int(100 * (float(written) / float(total)))

        # 打印当前下载进度，\r 表示回到行首，实现命令行中实时刷新效果
        # end='' 表示不换行，使下一次打印覆盖当前行
        print(f'\r{rate}% ', end='')

    # 执行获取对象的请求，指定存储空间名称、对象名称及进度回调函数
    result = client.get_object_to_file(
        oss.GetObjectRequest(
            bucket=args.bucket,  # 指定存储空间名称
            key=args.key,        # 指定对象键名
            progress_fn=_progress_fn, # 指定进度回调函数
        ),
        "/local/dir/example", # 指定保存到本地的文件路径
    )

    # 输出获取对象的结果信息
    print(vars(result))

# 当此脚本被直接执行时，调用main函数开始处理逻辑
if __name__ == "__main__":
    main()  # 脚本入口点，控制程序流程从这里开始

以下示例代码展示了在流式下载时，使用进度条查看下载文件的进度，以get_object为例。

import argparse
import alibabacloud_oss_v2 as oss
import os

# 创建命令行参数解析器
parser = argparse.ArgumentParser(description="get object sample")

# 添加命令行参数 --region，表示存储空间所在的区域，必需参数
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# 添加命令行参数 --bucket，表示存储空间的名称，必需参数
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# 添加命令行参数 --endpoint，表示其他服务可用来访问OSS的域名，非必需参数
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
# 添加命令行参数 --key，表示对象的名称，必需参数
parser.add_argument('--key', help='The name of the object.', required=True)

def main():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载访问OSS所需的认证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载SDK的默认配置，并设置凭证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 设置配置中的区域信息
    cfg.region = args.region

    # 如果提供了endpoint参数，则设置配置中的endpoint
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置好的信息创建OSS客户端
    client = oss.Client(cfg)

    # 执行获取对象的请求，指定存储空间名称和对象名称
    result = client.get_object(oss.GetObjectRequest(
        bucket=args.bucket,  # 指定存储空间名称
        key=args.key,  # 指定对象键名
    ))

    # 获取对象返回的结果中包含文件的总大小（字节数）
    total_size = result.content_length

    # 初始化进度计数器为 0，用于记录已下载的数据量
    progress_save_n = 0

    # 遍历响应体中的数据块，实现逐块读取数据
    for d in result.body.iter_bytes():
        # 累加当前数据块的长度到已下载总量中
        progress_save_n += len(d)

        # 计算当前下载百分比，将已下载量与总大小进行比例换算并取整
        rate = int(100 * (float(progress_save_n) / float(total_size)))

        # 打印当前下载进度，\r 表示回到行首，实现命令行中实时刷新效果
        # end='' 表示不换行，使下一次打印覆盖当前行
        print(f'\r{rate}% ', end='')

    # 打印结果对象的所有属性信息，用于调试或查看完整响应内容
    print(vars(result))


# 当此脚本被直接运行时，调用main函数
if __name__ == "__main__":
    main()  # 脚本入口，当文件被直接运行时调用main函数

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import argparse
import alibabacloud_oss_v2 as oss
import os
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import List, Tuple, Optional
import signal

class DownloadTask:
    """下载任务类"""
    def __init__(self, object_key: str, local_path: str, size: int):
        self.object_key = object_key
        self.local_path = local_path
        self.size = size

class DownloadResult:
    """下载结果类"""
    def __init__(self, object_key: str, success: bool = False, error: Optional[str] = None, size: int = 0):
        self.object_key = object_key
        self.success = success
        self.error = error
        self.size = size

class BatchDownloader:
    """批量下载器"""

    def __init__(self, client: oss.Client, bucket: str, max_workers: int = 5):
        self.client = client
        self.bucket = bucket
        self.max_workers = max_workers
        self.stop_event = threading.Event()

    def list_objects(self, prefix: str = "", max_keys: int = 1000) -> List[DownloadTask]:
        """列举存储空间中指定前缀的所有对象"""
        tasks = []
        continuation_token = None

        print(f"正在扫描存储空间中的文件...")

        while not self.stop_event.is_set():
            try:
                # 创建列举对象请求
                request = oss.ListObjectsV2Request(
                    bucket=self.bucket,
                    prefix=prefix,
                    max_keys=max_keys,
                    continuation_token=continuation_token
                )

                # 执行列举操作
                result = self.client.list_objects_v2(request)

                # 处理列举结果
                for obj in result.contents:
                    # 跳过文件夹对象（以/结尾且大小为0）
                    if obj.key.endswith('/') and obj.size == 0:
                        continue

                    # 计算本地文件路径
                    relative_path = obj.key[len(prefix):] if prefix else obj.key

                    tasks.append(DownloadTask(
                        object_key=obj.key,
                        local_path=relative_path,
                        size=obj.size
                    ))

                # 检查是否还有更多对象
                if not result.next_continuation_token:
                    break
                continuation_token = result.next_continuation_token

            except Exception as e:
                raise Exception(f"列举对象失败: {str(e)}")

        return tasks

    def download_file(self, task: DownloadTask, local_dir: str) -> DownloadResult:
        """下载单个文件"""
        result = DownloadResult(task.object_key, size=task.size)

        try:
            # 计算完整的本地文件路径
            full_local_path = os.path.join(local_dir, task.local_path)

            # 创建本地文件目录
            os.makedirs(os.path.dirname(full_local_path), exist_ok=True)

            # 检查文件是否已存在且大小一致（断点续传）
            if os.path.exists(full_local_path):
                local_size = os.path.getsize(full_local_path)
                if local_size == task.size:
                    result.success = True
                    return result

            # 创建下载请求
            get_request = oss.GetObjectRequest(
                bucket=self.bucket,
                key=task.object_key
            )

            # 执行下载
            response = self.client.get_object(get_request)

            # 保存文件
            with open(full_local_path, 'wb') as f:
                with response.body as body_stream:
                    # 分块读取并写入
                    for chunk in body_stream.iter_bytes(block_size=1024 * 1024):  # 1MB块
                        if self.stop_event.is_set():
                            raise Exception("下载被中断")
                        f.write(chunk)

            result.success = True

        except Exception as e:
            result.error = str(e)
            # 如果下载失败，删除不完整的文件
            try:
                if os.path.exists(full_local_path):
                    os.remove(full_local_path)
            except:
                pass

        return result

    def batch_download(self, tasks: List[DownloadTask], local_dir: str) -> List[DownloadResult]:
        """执行批量下载"""
        results = []
        completed = 0
        total = len(tasks)

        print(f"开始下载 {total} 个文件，使用 {self.max_workers} 个并发...")

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # 提交所有下载任务
            future_to_task = {
                executor.submit(self.download_file, task, local_dir): task
                for task in tasks
            }

            # 处理完成的任务
            for future in as_completed(future_to_task):
                if self.stop_event.is_set():
                    break

                task = future_to_task[future]
                try:
                    result = future.result()
                    results.append(result)
                    completed += 1

                    # 显示进度
                    if result.success:
                        print(f"✓ [{completed}/{total}] {result.object_key} ({self.format_bytes(result.size)})")
                    else:
                        print(f"✗ [{completed}/{total}] {result.object_key} - 错误: {result.error}")

                except Exception as e:
                    result = DownloadResult(task.object_key, error=str(e), size=task.size)
                    results.append(result)
                    completed += 1
                    print(f"✗ [{completed}/{total}] {task.object_key} - 异常: {str(e)}")

        return results

    def stop(self):
        """停止下载"""
        self.stop_event.set()
        print("\n正在停止下载...")

    @staticmethod
    def format_bytes(bytes_size: int) -> str:
        """格式化字节数为可读格式"""
        for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
            if bytes_size < 1024.0:
                return f"{bytes_size:.1f} {unit}"
            bytes_size /= 1024.0
        return f"{bytes_size:.1f} PB"

def signal_handler(signum, frame):
    """信号处理器"""
    print(f"\n接收到信号 {signum}，正在停止...")
    if hasattr(signal_handler, 'downloader'):
        signal_handler.downloader.stop()
    sys.exit(0)

def main():
    # 创建命令行参数解析器
    parser = argparse.ArgumentParser(description="OSS 批量下载工具")

    # 添加命令行参数
    parser.add_argument('--region', help='存储空间所在的区域', required=True)
    parser.add_argument('--bucket', help='存储空间的名称', required=True)
    parser.add_argument('--endpoint', help='自定义访问域名（可选）')
    parser.add_argument('--prefix', help='要下载的文件夹前缀，空字符串表示下载整个存储空间', default="")
    parser.add_argument('--local-dir', help='本地下载目录', default="./downloads")
    parser.add_argument('--workers', help='并发下载数量', type=int, default=5)
    parser.add_argument('--max-keys', help='每次列举的最大对象数', type=int, default=1000)

    # 解析命令行参数
    args = parser.parse_args()

    try:
        # 从环境变量中加载凭证信息，用于身份验证
        credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

        # 加载SDK的默认配置
        cfg = oss.config.load_default()
        cfg.credentials_provider = credentials_provider
        cfg.region = args.region

        # 如果提供了endpoint参数，则设置自定义endpoint
        if args.endpoint:
            cfg.endpoint = args.endpoint

        # 创建OSS客户端
        client = oss.Client(cfg)

        # 创建本地下载目录
        local_dir = getattr(args, 'local_dir')
        os.makedirs(local_dir, exist_ok=True)

        # 创建批量下载器
        downloader = BatchDownloader(client, args.bucket, args.workers)

        # 设置信号处理器以支持优雅停止
        signal_handler.downloader = downloader
        signal.signal(signal.SIGINT, signal_handler)
        signal.signal(signal.SIGTERM, signal_handler)

        print(f"开始批量下载")
        print(f"存储空间: {args.bucket}")
        print(f"前缀: '{args.prefix}' {'(整个存储空间)' if not args.prefix else ''}")
        print(f"本地目录: {local_dir}")
        print(f"并发数: {args.workers}")
        print("-" * 50)

        # 列举所有需要下载的对象
        tasks = downloader.list_objects(args.prefix, getattr(args, 'max_keys'))

        if not tasks:
            print("没有找到需要下载的文件")
            return

        print(f"找到 {len(tasks)} 个文件需要下载")
        print("-" * 50)

        # 执行批量下载
        start_time = time.time()
        results = downloader.batch_download(tasks, local_dir)
        end_time = time.time()

        # 统计下载结果
        success_count = sum(1 for r in results if r.success)
        fail_count = len(results) - success_count
        total_size = sum(r.size for r in results if r.success)
        duration = end_time - start_time

        print("-" * 50)
        print(f"下载完成!")
        print(f"成功: {success_count}")
        print(f"失败: {fail_count}")
        print(f"总大小: {BatchDownloader.format_bytes(total_size)}")
        print(f"耗时: {duration:.2f} 秒")

        if fail_count > 0:
            print(f"\n失败的文件:")
            for result in results:
                if not result.success:
                    print(f"  - {result.object_key}: {result.error}")

    except KeyboardInterrupt:
        print("\n下载被用户中断")
        sys.exit(1)
    except Exception as e:
        print(f"错误: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()

使用示例

# 下载 my-bucket 中 images/2024/ 文件夹下的所有文件
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix images/2024/

# 下载到指定本地目录
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix documents/ --local-dir ./my-downloads

# 使用更多并发下载
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix videos/ --workers 10

# 下载整个存储空间（不指定prefix或使用空字符串）
python batch_download.py --region cn-hangzhou --bucket my-bucket

# 或者显式指定空前缀
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix ""

输出示例

程序运行时会显示详细的下载进度：

开始批量下载
存储空间: my-bucket
前缀: 'images/2024/'
本地目录: ./downloads
并发数: 5
--------------------------------------------------
正在扫描存储空间中的文件...
找到 150 个文件需要下载
--------------------------------------------------
开始下载 150 个文件，使用 5 个并发...
✓ [1/150] images/2024/photo1.jpg (2.3 MB)
✓ [2/150] images/2024/photo2.png (1.8 MB)
✗ [3/150] images/2024/photo3.gif - 错误: Request timeout
✓ [4/150] images/2024/subfolder/photo4.jpg (3.1 MB)
...
✓ [150/150] images/2024/thumbnails/thumb150.jpg (256.0 KB)
--------------------------------------------------
下载完成!
成功: 148
失败: 2
总大小: 1.2 GB
耗时: 45.67 秒

失败的文件:
  - images/2024/photo3.gif: Request timeout
  - images/2024/corrupted.jpg: Invalid response