Cara mengunduh objek dari bucket ke file lokal menggunakan Python SDK V2 - Object Storage Service

Topik ini menjelaskan cara dengan cepat mengunduh objek dari bucket Object Storage Service (OSS) ke perangkat lokal.

Catatan penggunaan

Kode contoh dalam topik ini menggunakan ID wilayah cn-hangzhou dari wilayah Hangzhou, Tiongkok. Secara default, titik akhir publik digunakan untuk mengakses sumber daya di bucket. Jika Anda ingin mengakses sumber daya di bucket dari layanan Alibaba Cloud lainnya di wilayah yang sama dengan bucket tersebut, gunakan titik akhir internal. Untuk informasi lebih lanjut tentang wilayah dan titik akhir yang didukung oleh OSS, lihat Wilayah dan Titik Akhir.

Izin

Secara default, akun Alibaba Cloud memiliki izin penuh. Pengguna RAM atau Peran RAM di bawah akun Alibaba Cloud tidak memiliki izin apa pun secara default. Akun Alibaba Cloud atau administrator akun harus memberikan izin operasi melalui Kebijakan RAM atau Kebijakan Bucket.

API	Aksi	Definisi
GetObject	`oss:GetObject`	Mengunduh objek.
	`oss:GetObjectVersion`	Saat mengunduh objek, jika Anda menentukan versi objek melalui versionId, izin ini diperlukan.
	`kms:Decrypt`	Saat mengunduh objek, jika metadata objek berisi X-Oss-Server-Side-Encryption: KMS, izin ini diperlukan.

Definisi metode

get_object(request: GetObjectRequest, **kwargs) → GetObjectResult

Parameter permintaan

Parameter	Tipe	Deskripsi
request	GetObjectRequest	Parameter permintaan. Untuk informasi lebih lanjut, lihat GetObjectRequest

Parameter respons

Tipe	Deskripsi
GetObjectResult	Nilai kembali. Untuk detailnya, lihat GetObjectResult

Untuk definisi lengkap metode unduhan sederhana, lihat get_object.

Kode contoh

Kode contoh berikut mengunduh objek ke perangkat lokal:

import argparse
import alibabacloud_oss_v2 as oss
import os

# Buat parser parameter baris perintah.
parser = argparse.ArgumentParser(description="contoh unduh objek")

# Tentukan parameter --region untuk menunjukkan wilayah tempat bucket berada. Parameter ini wajib.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
# Tentukan parameter --bucket untuk menunjukkan nama bucket. Parameter baris perintah ini wajib. 
parser.add_argument('--bucket', help='Nama bucket.', required=True)
# Tentukan parameter --endpoint untuk menunjukkan titik akhir wilayah tempat bucket berada. Parameter ini opsional.
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')
# Tentukan parameter --key untuk menunjukkan nama objek. Parameter ini wajib.
parser.add_argument('--key', help='Nama objek.', required=True)

def main():
    # Parse parameter baris perintah.
    args = parser.parse_args()

    # Dapatkan kredensial akses dari variabel lingkungan untuk otentikasi.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Muat konfigurasi default SDK dan tentukan penyedia kredensial.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Tentukan wilayah tempat bucket berada.
    cfg.region = args.region

    # Jika titik akhir disediakan, tentukan titik akhir dalam objek konfigurasi.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Gunakan konfigurasi untuk membuat instance OSSClient.
    client = oss.Client(cfg)

    # Eksekusi permintaan untuk mengunduh objek, dan tentukan nama bucket dan nama objek.
    result = client.get_object(oss.GetObjectRequest(
        bucket=args.bucket,  # Tentukan nama bucket.
        key=args.key,  # Tentukan kunci objek.
    ))

    # Tampilkan respons untuk memeriksa apakah permintaan berhasil.
    print(f'kode status: {result.status_code},'
          f' ID permintaan: {result.request_id},'
          f' panjang konten: {result.content_length},'
          f' rentang konten: {result.content_range},'
          f' tipe konten: {result.content_type},'
          f' etag: {result.etag},'
          f' terakhir dimodifikasi: {result.last_modified},'
          f' MD5 konten: {result.content_md5},'
          f' kontrol cache: {result.cache_control},'
          f' disposisi konten: {result.content_disposition},'
          f' pengkodean konten: {result.content_encoding},'
          f' kedaluwarsa: {result.expires},'
          f' hash crc64: {result.hash_crc64},'
          f' kelas penyimpanan: {result.storage_class},'
          f' tipe objek: {result.object_type},'
          f' ID versi: {result.version_id},'
          f' jumlah tagging: {result.tagging_count},'
          f' enkripsi sisi server: {result.server_side_encryption},'
          f' enkripsi data sisi server: {result.server_side_data_encryption},'
          f' posisi tambahan berikutnya: {result.next_append_position},'
          f' kedaluwarsa: {result.expiration},'
          f' pemulihan: {result.restore},'
          f' status proses: {result.process_status},'
          f' penanda hapus: {result.delete_marker},'
    )

    # ========== Metode 1: Baca seluruh objek ==========
    with result.body as body_stream:
        data = body_stream.read()
        print(f"Objek telah dibaca. Panjang data: {len(data)} byte")

        path = "./get-object-sample.txt"
        with open(path, 'wb') as f:
            f.write(data)
        print(f"Objek telah diunduh dan disimpan ke jalur lokal: {path}")

    # # ========== Metode 2: Baca objek dalam blok ==========
    # with result.body as body_stream:
    #     chunk_path = "./get-object-sample-chunks.txt"
    #     total_size = 0

    #     with open(chunk_path, 'wb') as f:
    #         # Gunakan ukuran blok 256KB (Anda dapat menyesuaikan parameter block_size sesuai kebutuhan)
    #         for chunk in body_stream.iter_bytes(block_size=256 * 1024):
    #             f.write(chunk)
    #             total_size += len(chunk)
    #             print(f"Blok data diterima: {len(chunk)} byte | Total: {total_size} byte")

    #     print(f"Objek telah diunduh dan disimpan ke jalur lokal: {chunk_path}")

# Panggil fungsi utama saat skrip dijalankan langsung.
if __name__ == "__main__":
    main()  # Titik masuk skrip. Saat skrip dijalankan langsung, fungsi utama dipanggil.

Skenario umum

Unduhan bersyarat

Saat Anda mengunduh satu objek dari bucket, Anda dapat menentukan kondisi berdasarkan waktu modifikasi terakhir atau ETag objek. Objek hanya akan diunduh jika kondisi-kondisi ini terpenuhi. Jika tidak, kesalahan akan dikembalikan, dan operasi unduhan tidak akan dipicu. Ini mengurangi transmisi jaringan dan konsumsi sumber daya yang tidak perlu, serta meningkatkan efisiensi unduhan.

Tabel berikut menjelaskan kondisi yang tersedia.

Catatan

if_modified_since dan if_unmodified_since dapat ada bersamaan. if_match dan if_none_match juga dapat ada bersamaan.
Anda dapat menggunakan metode client.get_object_meta untuk mendapatkan ETag.

Parameter	Deskripsi
if_modified_since	Jika waktu yang ditentukan lebih awal daripada waktu terakhir kali objek dimodifikasi, objek dapat diunduh. Jika tidak, 304 Not Modified dikembalikan.
if_unmodified_since	Jika waktu yang ditentukan lebih lambat atau sama dengan waktu terakhir kali objek dimodifikasi, objek dapat diunduh. Jika tidak, 412 Precondition Failed dikembalikan.
if_match	Jika ETag yang ditentukan cocok dengan ETag objek, objek dapat diunduh. Jika tidak, 412 Precondition Failed dikembalikan.
if_none_match	Jika ETag yang ditentukan tidak cocok dengan ETag objek, objek dapat diunduh. Jika tidak, 304 Not Modified dikembalikan.

Kode contoh berikut memberikan contoh cara menerapkan unduhan bersyarat:

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

# Buat parser untuk mem-parsing argumen baris perintah dan jelaskan tujuan skrip.
parser = argparse.ArgumentParser(description="unduh objek ke file sampel")

# Tentukan parameter --region untuk menunjukkan wilayah tempat bucket berada. Parameter ini wajib.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
# Tentukan parameter --bucket untuk menunjukkan nama bucket tempat objek disimpan. Parameter ini wajib.
parser.add_argument('--bucket', help='Nama bucket.', required=True)
# Tentukan parameter --endpoint untuk menunjukkan titik akhir wilayah tempat bucket berada. Parameter ini opsional.
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')
# Tentukan parameter --key untuk menunjukkan nama objek. Parameter ini wajib.
parser.add_argument('--key', help='Nama objek.', required=True)
# Tentukan parameter --file_path untuk menunjukkan jalur tempat objek diunduh. Parameter ini wajib.
parser.add_argument('--file_path', help='Jalur file untuk menyimpan konten yang diunduh.', required=True)

def main():
    # Parse parameter baris perintah untuk mendapatkan nilai-nilai yang ditentukan.
    args = parser.parse_args()

    # Dapatkan kredensial akses dari variabel lingkungan untuk otentikasi.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Gunakan konfigurasi default untuk membuat objek konfigurasi (cfg) dan tentukan penyedia kredensial.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    
    # Set atribut region objek cfg ke region di parser.
    cfg.region = args.region

    # Jika titik akhir kustom disediakan, perbarui atribut endpoint objek cfg dengan titik akhir yang diberikan.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Gunakan konfigurasi sebelumnya untuk menginisialisasi instance OSSClient.
    client = oss.Client(cfg)

    # Definisikan waktu if_modified_since.
    # Hanya objek yang dimodifikasi setelah waktu ini yang akan dikembalikan.
    if_modified_since = datetime(2024, 10, 1, 12, 0, 0, tzinfo=timezone.utc)

    # Asumsikan ETag adalah DA5223EFCD7E0353BE08866700000000, jika ETag yang ditentukan cocok dengan ETag objek, kondisi IfMatch terpenuhi dan unduhan akan dipicu.
    etag = "\"DA5223EFCD7E0353BE08866700000000\""

    # Eksekusi permintaan untuk mengunduh objek dan menyimpannya ke file lokal.
    result = client.get_object_to_file(
        oss.GetObjectRequest(
            bucket=args.bucket,  # Tentukan nama bucket.
            key=args.key,        # Tentukan kunci objek.
            if_modified_since=if_modified_since,  # Hanya objek yang dimodifikasi setelah waktu yang ditentukan yang akan dikembalikan.
            if_match=etag,       # Hanya objek dengan ETag yang cocok yang akan dikembalikan.
        ),
        args.file_path  # Tentukan jalur lokal untuk menyimpan file yang diunduh.
    )

    # Tampilkan informasi respons, termasuk kode status, ID permintaan, dll.
    print(f'kode status: {result.status_code},'
          f' ID permintaan: {result.request_id},'
          f' panjang konten: {result.content_length},'
          f' rentang konten: {result.content_range},'
          f' tipe konten: {result.content_type},'
          f' etag: {result.etag},'
          f' terakhir dimodifikasi: {result.last_modified},'
          f' MD5 konten: {result.content_md5},'
          f' kontrol cache: {result.cache_control},'
          f' disposisi konten: {result.content_disposition},'
          f' pengkodean konten: {result.content_encoding},'
          f' kedaluwarsa: {result.expires},'
          f' hash crc64: {result.hash_crc64},'
          f' kelas penyimpanan: {result.storage_class},'
          f' tipe objek: {result.object_type},'
          f' ID versi: {result.version_id},'
          f' jumlah tagging: {result.tagging_count},'
          f' enkripsi sisi server: {result.server_side_encryption},'
          f' enkripsi data sisi server: {result.server_side_data_encryption},'
          f' posisi tambahan berikutnya: {result.next_append_position},'
          f' kedaluwarsa: {result.expiration},'
          f' pemulihan: {result.restore},'
          f' status proses: {result.process_status},'
          f' penanda hapus: {result.delete_marker},'
          f' waktu server: {result.headers.get("x-oss-server-time")},'
    )

# Panggil fungsi utama untuk memulai logika pemrosesan saat skrip dijalankan langsung.
if __name__ == "__main__":
    main()  # Tentukan titik masuk skrip. Aliran kontrol dimulai di sini.

Tampilkan kemajuan unduhan

Saat Anda mengunduh objek, Anda dapat menggunakan bilah kemajuan untuk memantau kemajuan unduhan secara real time. Pemantauan kemajuan membantu Anda memeriksa apakah tugas unduhan macet jika memerlukan waktu lama untuk diselesaikan.

Kode contoh berikut menunjukkan cara menggunakan bilah kemajuan untuk melihat kemajuan unduhan saat mengunduh objek ke file lokal menggunakan get_object_to_file.

import argparse
import alibabacloud_oss_v2 as oss

# Buat parser untuk mem-parsing argumen baris perintah dan jelaskan tujuan skrip.
parser = argparse.ArgumentParser(description="contoh unduh objek")

# Tentukan parameter --region untuk menunjukkan wilayah tempat bucket berada. Parameter ini wajib.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
# Tentukan parameter --bucket untuk menunjukkan nama bucket tempat objek disimpan. Parameter ini wajib.
parser.add_argument('--bucket', help='Nama bucket.', required=True)
# Tentukan parameter --endpoint untuk menunjukkan titik akhir wilayah tempat bucket berada. Parameter ini opsional.
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')
# Tentukan parameter --key untuk menunjukkan nama objek. Parameter ini wajib. 
parser.add_argument('--key', help='Nama objek.', required=True)

def main():
    # Parse parameter baris perintah untuk mendapatkan nilai-nilai yang ditentukan.
    args = parser.parse_args()

    # Dapatkan kredensial akses dari variabel lingkungan untuk otentikasi.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Gunakan konfigurasi default untuk membuat objek konfigurasi (cfg) dan tentukan penyedia kredensial.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Set atribut region objek cfg ke region di parser.
    cfg.region = args.region

    # Jika titik akhir kustom disediakan, perbarui atribut endpoint objek cfg dengan titik akhir yang diberikan.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Gunakan konfigurasi sebelumnya untuk menginisialisasi instance OSSClient.
    client = oss.Client(cfg)

    # Definisikan variabel kamus progress_state untuk menyimpan status kemajuan unduhan, dengan nilai awal 0.
    progress_state = {'saved': 0}
    
    # Definisikan fungsi callback kemajuan _progress_fn.
    def _progress_fn(n, written, total):
        # Gunakan kamus untuk menyimpan jumlah byte tertulis yang terakumulasi.
        progress_state['saved'] += n

        # Hitung persentase unduhan saat ini dengan membagi jumlah byte tertulis dengan total byte dan memotong hasilnya menjadi integer.
        rate = int(100 * (float(written) / float(total)))

        # Cetak kemajuan unduhan saat ini. Gunakan \r untuk kembali ke awal baris, mencapai pembaruan real time di terminal.
        # end='' menekan pemutusan baris, memungkinkan cetakan berikutnya menimpa baris saat ini.
        print(f'\r{rate}% ', end='')

    # Eksekusi permintaan untuk mengunduh objek, dan tentukan nama bucket, nama objek, dan fungsi callback kemajuan.
    result = client.get_object_to_file(
        oss.GetObjectRequest(
            bucket=args.bucket,  # Tentukan nama bucket.
            key=args.key,        # Tentukan kunci objek.
            progress_fn=_progress_fn, # Tentukan fungsi callback kemajuan.
        ),
        "/local/dir/contoh", # Tentukan jalur lokal untuk menyimpan file.
    )

    # Tampilkan informasi respons.
    print(vars(result))

# Panggil fungsi utama untuk memulai logika pemrosesan saat skrip dijalankan langsung.
if __name__ == "__main__":
    main()  # Tentukan titik masuk skrip. Aliran kontrol dimulai di sini.

Kode contoh berikut menunjukkan cara menggunakan bilah kemajuan untuk melihat kemajuan unduhan saat melakukan streaming objek menggunakan get_object.

import argparse
import alibabacloud_oss_v2 as oss
import os

# Buat parser parameter baris perintah.
parser = argparse.ArgumentParser(description="contoh unduh objek")

# Tentukan parameter --region untuk menunjukkan wilayah tempat bucket berada. Parameter ini wajib.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
# Tentukan parameter --bucket untuk menunjukkan nama bucket. Parameter baris perintah ini wajib.
parser.add_argument('--bucket', help='Nama bucket.', required=True)
# Tentukan parameter --endpoint untuk menunjukkan titik akhir wilayah tempat bucket berada. Parameter ini opsional.
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')
# Tentukan parameter --key untuk menunjukkan nama objek. Parameter ini wajib.
parser.add_argument('--key', help='Nama objek.', required=True)

def main():
    # Parse parameter baris perintah.
    args = parser.parse_args()

    # Dapatkan kredensial akses dari variabel lingkungan untuk otentikasi.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Muat konfigurasi default SDK dan tentukan penyedia kredensial.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Tentukan wilayah tempat bucket berada.
    cfg.region = args.region

    # Jika titik akhir disediakan, tentukan titik akhir dalam objek konfigurasi.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Gunakan konfigurasi untuk membuat instance OSSClient.
    client = oss.Client(cfg)

    # Eksekusi permintaan untuk mengunduh objek, dan tentukan nama bucket dan nama objek.
    result = client.get_object(oss.GetObjectRequest(
        bucket=args.bucket,  # Tentukan nama bucket.
        key=args.key,  # Tentukan kunci objek.
    ))

    # Dapatkan ukuran total file dalam byte dari hasil respons objek.
    total_size = result.content_length

    # Inisialisasi penghitung kemajuan ke 0, digunakan untuk mencatat jumlah data yang diunduh.
    progress_save_n = 0

    # Iterasi melalui blok data dalam respons body untuk menerapkan pembacaan data blok demi blok.
    for d in result.body.iter_bytes():
        # Akumulasikan panjang blok data saat ini ke jumlah total yang diunduh.
        progress_save_n += len(d)

        # Hitung persentase unduhan saat ini dengan menghitung rasio jumlah yang diunduh terhadap ukuran total dan membulatkannya ke integer terdekat.
        rate = int(100 * (float(progress_save_n) / float(total_size)))

        # Cetak kemajuan unduhan saat ini. Gunakan \r (carriage return) untuk memposisikan ulang kursor di awal baris, memungkinkan pembaruan kemajuan real time dalam antarmuka baris perintah.
        # end='' menekan pemutusan baris, memungkinkan cetakan berikutnya menimpa baris saat ini.
        print(f'\r{rate}% ', end='')

    # Cetak semua atribut objek hasil untuk tujuan debugging atau untuk memeriksa isi respons lengkap.
    print(vars(result))


# Panggil fungsi utama untuk memulai logika pemrosesan saat skrip dijalankan langsung.
if __name__ == "__main__":
    main()  # Tentukan titik masuk skrip. Aliran kontrol dimulai di sini.

Unduh objek secara batch ke file lokal

#!/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:
    """Kelas tugas unduhan"""
    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:
    """Kelas hasil unduhan"""
    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:
    """Pengunduh batch"""

    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]:
        """Daftar semua objek dengan awalan tertentu di bucket"""
        tasks = []
        continuation_token = None

        print(f"Mencari file di bucket...")

        while not self.stop_event.is_set():
            try:
                # Buat permintaan untuk mencatat objek。
                request = oss.ListObjectsV2Request(
                    bucket=self.bucket,
                    prefix=prefix,
                    max_keys=max_keys,
                    continuation_token=continuation_token
                )

                # Eksekusi operasi daftar.
                result = self.client.list_objects_v2(request)

                # Proses hasil daftar.
                for obj in result.contents:
                    # Lewati objek folder (diidentifikasi oleh akhiran '/' dan ukuran 0-byte).
                    if obj.key.endswith('/') and obj.size == 0:
                        continue

                    # Hitung jalur file lokal.
                    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
                    ))

                # Periksa apakah ada lebih banyak objek.
                if not result.next_continuation_token:
                    break
                continuation_token = result.next_continuation_token

            except Exception as e:
                raise Exception(f"Gagal mencatat objek: {str(e)}")

        return tasks

    def download_file(self, task: DownloadTask, local_dir: str) -> DownloadResult:
        """Unduh file tunggal"""
        result = DownloadResult(task.object_key, size=task.size)

        try:
            # Hitung jalur file lokal penuh.
            full_local_path = os.path.join(local_dir, task.local_path)

            # Buat direktori file lokal.
            os.makedirs(os.path.dirname(full_local_path), exist_ok=True)

            # Verifikasi keberadaan file dan validasi konsistensi ukuran untuk memungkinkan transfer yang dapat dilanjutkan.
            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

            # Buat permintaan unduhan.
            get_request = oss.GetObjectRequest(
                bucket=self.bucket,
                key=task.object_key
            )

            # Eksekusi operasi unduhan.
            response = self.client.get_object(get_request)

            # Simpan file.
            with open(full_local_path, 'wb') as f:
                with response.body as body_stream:
                    # Baca dan tulis dalam blok
                    for chunk in body_stream.iter_bytes(block_size=1024 * 1024):  # Blok 1MB
                        if self.stop_event.is_set():
                            raise Exception("Unduhan terganggu")
                        f.write(chunk)

            resultsuccess = True

        except Exception as e:
            result.error = str(e)
            # Jika unduhan gagal, hapus file yang tidak lengkap.
            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]:
        """Eksekusi unduhan batch"""
        results = []
        completed = 0
        total = len(tasks)

        print(f"Memulai mengunduh {total} file menggunakan {self.max_workers} thread konkuren...")

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # Kirim semua tugas unduhan.
            future_to_task = {
                executor.submit(self.download_file, task, local_dir): task
                for task in tasks
            }

            # Proses tugas yang selesai.
            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

                    # Tampilkan kemajuan.
                    if result.success:
                        print(f"✓ [{completed}/{total}] {result.object_key} ({self.format_bytes(result.size)})")
                    else:
                        print(f"✗ [{completed}/{total}] {result.object_key} - Kesalahan: {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} - Pengecualian: {str(e)}")

        return results

    def stop(self):
        """Hentikan unduhan"""
        self.stop_event.set()
        print("\nMenghentikan unduhan...")

    @staticmethod
    def format_bytes(bytes_size: int) -> str:
        """Format byte ke format yang dapat dibaca"""
        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):
    """Penangan sinyal"""
    print(f"\nMenerima sinyal {signum}, menghentikan...")
    if hasattr(signal_handler, 'downloader'):
        signal_handler.downloader.stop()
    sys.exit(0)

def main():
    # Buat parser parameter baris perintah.
    parser = argparse.ArgumentParser(description="Alat Unduhan Batch OSS")

    # Tambahkan parameter baris perintah.
    parser.add_argument('--region', help='Wilayah tempat bucket berada', required=True)
    parser.add_argument('--bucket', help='Nama bucket', required=True)
    parser.add_argument('--endpoint', help='Titik akhir kustom (opsional)')
    parser.add_argument('--prefix', help='Awalan folder untuk diunduh, string kosong berarti unduh seluruh bucket', default="")
    parser.add_argument('--local-dir', help='Direktori unduhan lokal', default="./downloads")
    parser.add_argument('--workers', help='Jumlah unduhan konkuren', type=int, default=5)
    parser.add_argument('--max-keys', help='Jumlah maksimum objek untuk dicatat sekaligus', type=int, default=1000)

    # Parse parameter baris perintah.
    args = parser.parse_args()

    try:
        # Dapatkan kredensial akses dari variabel lingkungan untuk otentikasi.
        credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

        # Muat konfigurasi default SDK.
        cfg = oss.config.load_default()
        cfg.credentials_provider = credentials_provider
        cfg.region = args.region

        # Jika parameter titik akhir disediakan, tentukan titik akhir kustom.
        if args.endpoint:
            cfg.endpoint = args.endpoint

        # Buat klien OSS.
        client = oss.Client(cfg)

        # Buat direktori unduhan lokal.
        local_dir = getattr(args, 'local_dir')
        os.makedirs(local_dir, exist_ok=True)

        # Buat pengunduh batch.
        downloader = BatchDownloader(client, args.bucket, args.workers)

        # Konfigurasikan penangan sinyal untuk memungkinkan prosedur penutupan yang aman.
        signal_handler.downloader = downloader
        signal.signal(signal.SIGINT, signal_handler)
        signal.signal(signal.SIGTERM, signal_handler)

        print(f"Memulai unduhan batch")
        print(f"Bucket: {args.bucket}")
        print(f"Awalan: '{args.prefix}' {'(seluruh bucket)' if not args.prefix else ''}")
        print(f"Direktori lokal: {local_dir}")
        print(f"Konkurensi: {args.workers}")
        print("-" * 50)

        # Daftar semua objek untuk diunduh.
        tasks = downloader.list_objects(args.prefix, getattr(args, 'max_keys'))

        if not tasks:
            print("Tidak ada file yang ditemukan untuk diunduh")
            return

        print(f"Ditemukan {len(tasks)} file untuk diunduh")
        print("-" * 50)

        # Eksekusi unduhan batch.
        start_time = time.time()
        results = downloader.batch_download(tasks, local_dir)
        end_time = time.time()

        # Rangkum hasil unduhan.
        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"Unduhan selesai!")
        print(f"Berhasil: {success_count}")
        print(f"Gagal: {fail_count}")
        print(f"Ukuran total: {BatchDownloader.format_bytes(total_size)}")
        print(f"Durasi: {duration:.2f} detik")

        if fail_count > 0:
            print(f"\nFile yang gagal:")
            for result in results:
                if not result.success:
                    print(f"  - {result.object_key}: {result.error}")

    except KeyboardInterrupt:
        print("\nUnduhan dihentikan oleh pengguna")
        sys.exit(1)
    except Exception as e:
        print(f"Kesalahan: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Contoh penggunaan

# Unduh semua objek dengan awalan images/2024/ dari bucket my-bucket.
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix images/2024/

# Unduh ke direktori lokal tertentu.
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix documents/ --local-dir ./my-downloads

# Tingkatkan konkurensi unduhan untuk meningkatkan throughput.
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix videos/ --workers 10

# Unduh semua objek di bucket dengan menghilangkan parameter awalan atau menentukan string kosong sebagai awalan.
python batch_download.py --region cn-hangzhou --bucket my-bucket

# Atau, tentukan awalan kosong secara eksplisit。
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix ""

Contoh keluaran

Program menampilkan kemajuan unduhan terperinci selama eksekusi:

Memulai unduhan batch
Bucket: my-bucket
Awalan: 'images/2024/'
Direktori lokal: ./downloads
Konkurensi: 5
--------------------------------------------------
Mencari file di bucket...
Ditemukan 150 file untuk diunduh
--------------------------------------------------
Memulai mengunduh 150 file menggunakan 5 thread konkuren...
✓ [1/150] images/2024/photo1.jpg (2.3 MB)
✓ [2/150] images/2024/photo2.png (1.8 MB)
✗ [3/150] images/2024/photo3.gif - Kesalahan: Waktu permintaan habis
✓ [4/150] images/2024/subfolder/photo4.jpg (3.1 MB)
...
✓ [150/150] images/2024/thumbnails/thumb150.jpg (256.0 KB)
--------------------------------------------------
Unduhan selesai!
Berhasil: 148
Gagal: 2
Ukuran total: 1.2 GB
Durasi: 45.67 detik

File yang gagal:
  - images/2024/photo3.gif: Waktu permintaan habis
  - images/2024/corrupted.jpg: Respons tidak valid

Referensi

Untuk kode contoh lengkap untuk unduhan sederhana, lihat get_object.py dan get_object_to_file.py.