Object Storage Service:Download simples (Python SDK V2)

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

# Create a command-line argument parser and describe the purpose of the script: get object and save to file sample.
parser = argparse.ArgumentParser(description="get object to file sample")

# Add the --region command-line argument to specify the region in which the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# Add the --bucket command-line argument to specify the name of the bucket from which to get the object. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# Add the --endpoint command-line argument to specify the domain name that other services can use to access OSS. This argument is not required.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
# Add the --key command-line argument to specify the key of the object (file) in OSS. This argument is required.
parser.add_argument('--key', help='The name of the object.', required=True)
# Add the --file_path command-line argument to specify the local path of the downloaded file. This argument is required.
parser.add_argument('--file_path', help='The path of the file to save the downloaded content.', required=True)

def main():
    # Parse the command-line arguments to get the user-input values.
    args = parser.parse_args()

    # Load the credentials required to access OSS from environment variables for identity verification.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Use the default configurations of the SDK to create a configuration object and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    
    # Set the region property of the configuration object based on the command-line arguments.
    cfg.region = args.region

    # If a custom endpoint is provided, update the endpoint property of the configuration object.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Use the preceding configurations to initialize the OSS client to interact with OSS.
    client = oss.Client(cfg)

    # Define the if_modified_since time.
    # Only objects modified after this time are returned.
    if_modified_since = datetime(2024, 10, 1, 12, 0, 0, tzinfo=timezone.utc)

    # Assume that the ETag is DA5223EFCD7E0353BE08866700000000. If the specified ETag is the same as the ETag of the object, the IfMatch condition is met and the download is triggered.
    etag = "\"DA5223EFCD7E0353BE08866700000000\""

    # Execute the request to get the object and save it to a local file.
    result = client.get_object_to_file(
        oss.GetObjectRequest(
            bucket=args.bucket,  # Specify the bucket name.
            key=args.key,        # Specify the object key.
            if_modified_since=if_modified_since,  # Only objects modified after the specified time are returned.
            if_match=etag,       # Only objects with a matching ETag are returned.
        ),
        args.file_path  # Specify the local path to which the file is downloaded.
    )

    # Print the result of getting the object, including the status code and request 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")},'
    )

# When this script is run directly, call the main function to start the processing logic.
if __name__ == "__main__":
    main()  # The entry point of the script. The program flow starts from here.

import argparse
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser and describe the purpose of the script: get object sample.
parser = argparse.ArgumentParser(description="get object sample")

# Add the --region command-line argument to specify the region in which the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# Add the --bucket command-line argument to specify the name of the bucket from which to get the object. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# Add the --endpoint command-line argument to specify the domain name that other services can use to access OSS. This argument is not required.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
# Add the --key command-line argument to specify the key of the object (file) in OSS. This argument is required.
parser.add_argument('--key', help='The name of the object.', required=True)

def main():
    # Parse the command-line arguments to get the user-input values.
    args = parser.parse_args()

    # Load the credentials required to access OSS from environment variables for identity verification.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Use the default configurations of the SDK to create a configuration object and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Set the region property of the configuration object based on the command-line arguments.
    cfg.region = args.region

    # If a custom endpoint is provided, update the endpoint property of the configuration object.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Use the preceding configurations to initialize the OSS client to interact with OSS.
    client = oss.Client(cfg)

    # Define a dictionary variable progress_state to save the download progress status. The initial value is 0.
    progress_state = {'saved': 0}
    
    # Define the progress callback function _progress_fn.
    def _progress_fn(n, written, total):
        # Use a dictionary to store the accumulated number of written bytes.
        progress_state['saved'] += n

        # Calculate the current download percentage. Divide the number of written bytes by the total number of bytes and round down the result to an integer.
        rate = int(100 * (float(written) / float(total)))

        # Print the current download progress. \r indicates returning to the beginning of the line to implement real-time refresh in the command line.
        # end='' indicates no line break, so that the next print overwrites the current line.
        print(f'\r{rate}% ', end='')

    # Execute the request to get the object. Specify the bucket name, object name, and progress callback function.
    result = client.get_object_to_file(
        oss.GetObjectRequest(
            bucket=args.bucket,  # Specify the bucket name.
            key=args.key,        # Specify the object key.
            progress_fn=_progress_fn, # Specify the progress callback function.
        ),
        "/local/dir/example", # Specify the local path to which the file is saved.
    )

    # Print the result of getting the object.
    print(vars(result))

# When this script is run directly, call the main function to start the processing logic.
if __name__ == "__main__":
    main()  # The entry point of the script. The program flow starts from here.

import argparse
import alibabacloud_oss_v2 as oss
import os

# Create a command-line argument parser.
parser = argparse.ArgumentParser(description="get object sample")

# Add the --region command-line argument to specify the region in which the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# Add the --bucket command-line argument to specify the name of the bucket. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# Add the --endpoint command-line argument to specify the domain name that other services can use to access OSS. This argument is not required.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
# Add the --key command-line argument to specify the name of the object. This argument is required.
parser.add_argument('--key', help='The name of the object.', required=True)

def main():
    # Parse the command-line arguments.
    args = parser.parse_args()

    # Load the credentials required to access OSS from environment variables for identity verification.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default configurations of the SDK and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Set the region in the configuration.
    cfg.region = args.region

    # If an endpoint is provided, set the endpoint in the configuration.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Use the configured information to create an OSS client.
    client = oss.Client(cfg)

    # Execute a request to get the object. Specify the bucket name and object name.
    result = client.get_object(oss.GetObjectRequest(
        bucket=args.bucket,  # Specify the bucket name.
        key=args.key,  # Specify the object key.
    ))

    # The result returned for getting the object contains the total size of the file in bytes.
    total_size = result.content_length

    # Initialize the progress counter to 0 to record the amount of downloaded data.
    progress_save_n = 0

    # Traverse the data blocks in the response body to read data block by block.
    for d in result.body.iter_bytes():
        # Add the length of the current data block to the total downloaded amount.
        progress_save_n += len(d)

        # Calculate the current download percentage. Convert the ratio of the downloaded amount to the total size to a percentage and round it down to an integer.
        rate = int(100 * (float(progress_save_n) / float(total_size)))

        # Print the current download progress. \r indicates returning to the beginning of the line to implement real-time refresh in the command line.
        # end='' indicates no line break, so that the next print overwrites the current line.
        print(f'\r{rate}% ', end='')

    # Print all property information of the result object for debugging or viewing the complete response content.
    print(vars(result))

# When this script is run directly, the main function is called.
if __name__ == "__main__":
    main()  # The entry point of the script. When the file is run directly, the main function is called.

#!/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:
    """Download task class"""
    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:
    """Download result class"""
    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:
    """Batch downloader"""

    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]:
        """List all objects that have the specified prefix in the bucket."""
        tasks = []
        continuation_token = None

        print(f"Scanning files in the bucket...")

        while not self.stop_event.is_set():
            try:
                # Create a request to list objects.
                request = oss.ListObjectsV2Request(
                    bucket=self.bucket,
                    prefix=prefix,
                    max_keys=max_keys,
                    continuation_token=continuation_token
                )

                # Execute the list operation.
                result = self.client.list_objects_v2(request)

                # Process the list result.
                for obj in result.contents:
                    # Skip folder objects, which end with a forward slash (/) and have a size of 0.
                    if obj.key.endswith('/') and obj.size == 0:
                        continue

                    # Calculate the local file path.
                    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
                    ))

                # Check whether there are more objects.
                if not result.next_continuation_token:
                    break
                continuation_token = result.next_continuation_token

            except Exception as e:
                raise Exception(f"Failed to list objects: {str(e)}")

        return tasks

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

        try:
            # Calculate the full local file path.
            full_local_path = os.path.join(local_dir, task.local_path)

            # Create the local file directory.
            os.makedirs(os.path.dirname(full_local_path), exist_ok=True)

            # Check whether the file exists and has the same size for resumable download.
            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

            # Create a download request.
            get_request = oss.GetObjectRequest(
                bucket=self.bucket,
                key=task.object_key
            )

            # Execute the download.
            response = self.client.get_object(get_request)

            # Save the file.
            with open(full_local_path, 'wb') as f:
                with response.body as body_stream:
                    # Read and write in chunks.
                    for chunk in body_stream.iter_bytes(block_size=1024 * 1024):  # 1 MB chunks
                        if self.stop_event.is_set():
                            raise Exception("Download interrupted")
                        f.write(chunk)

            result.success = True

        except Exception as e:
            result.error = str(e)
            # If the download fails, delete the incomplete file.
            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]:
        """Execute batch download."""
        results = []
        completed = 0
        total = len(tasks)

        print(f"Start to download {total} files using {self.max_workers} concurrent threads...")

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # Submit all download tasks.
            future_to_task = {
                executor.submit(self.download_file, task, local_dir): task
                for task in tasks
            }

            # Process completed 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

                    # Display the progress.
                    if result.success:
                        print(f"✓ [{completed}/{total}] {result.object_key} ({self.format_bytes(result.size)})")
                    else:
                        print(f"✗ [{completed}/{total}] {result.object_key} - Error: {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}]

# Download all files with the prefix images/2024/ from the my-bucket bucket.
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix images/2024/

# Download files to a specified local directory.
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix documents/ --local-dir ./my-downloads

# Use more concurrent workers for the download.
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix videos/ --workers 10

# Download all files in the bucket by either omitting the prefix parameter or specifying an empty string as the prefix.
python batch_download.py --region cn-hangzhou --bucket my-bucket

# Alternatively, explicitly specify an empty prefix.
python batch_download.py --region cn-hangzhou --bucket my-bucket --prefix ""

Starting batch download
Bucket: my-bucket
Prefix: 'images/2024/'
Local directory: ./downloads
Concurrency: 5
--------------------------------------------------
Scanning files in the bucket...
Found 150 files to download
--------------------------------------------------
Starting to download 150 files using 5 concurrent threads...
✓ [1/150] images/2024/photo1.jpg (2.3 MB)
✓ [2/150] images/2024/photo2.png (1.8 MB)
✗ [3/150] images/2024/photo3.gif - Error: Request timeout
✓ [4/150] images/2024/subfolder/photo4.jpg (3.1 MB)
...
✓ [150/150] images/2024/thumbnails/thumb150.jpg (256.0 KB)
--------------------------------------------------
Download complete!
Success: 148
Failed: 2
Total size: 1.2 GB
Duration: 45.67 seconds

Failed files:
  - images/2024/photo3.gif: Request timeout
  - images/2024/corrupted.jpg: Invalid response