Simple download (Python SDK V2)

When you download a single object from a bucket, you can specify conditions based on the last modified time or the ETag of the object. The object is downloaded only if these conditions are met. Otherwise, an error is returned, and the download operation is not triggered. This reduces unnecessary network transmission and resource consumption, and improves the download efficiency.

The following table describes the available conditions.

The following sample code shows how to use conditional download.

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.

When you download a file, you can use a progress bar to monitor the download progress in real time. This helps you track the download status and confirm that the task is proceeding as expected, which is especially useful for long-running downloads.

The following sample code shows how to display a progress bar when you download an object to a local file. The get_object_to_file method is used as an example.

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.

The following sample code shows how to display a progress bar for a streaming download. The get_object method is used as an example.

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}]

Usage examples

# 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 ""

Output example

The program displays detailed download progress at runtime:

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