Interacting with Aliyun Object Storage Service (OSS) requires authenticating the client and utilizing the bucket interface to perform CRUD operations on objects. The implementation below defines a class OSSManager to encapsulate these functionalities, including file transfers, streaming data, and generating temporary access URLs.

Client Initialization and Configuration

The class initializes by establishing an authenticated connection to the OSS bucket. It is best practice to retrieve sensitive credentials from environment variables rather than hardcoding them.

import os
import oss2

class OSSManager:
    def __init__(self, access_key_id, access_key_secret, endpoint, bucket_name):
        # Initialize authentication
        auth = oss2.Auth(access_key_id, access_key_secret)
        
        # Instantiate the bucket object for interaction
        self.bucket = oss2.Bucket(auth, endpoint, bucket_name)

    def check_object_exists(self, object_key):
        """Verify if a specific object exists in the bucket."""
        return self.bucket.object_exists(object_key)

Uploading Files

Files from the local filesystem can be uploaded to OSS. The following method allows uploading a file and optionally removing the local source after a successful transfer.

    def upload_file(self, local_path, remote_key=None, cleanup_source=False):
        # Use the local filename if a remote key is not specified
        target_key = remote_key or os.path.basename(local_path)
        
        self.bucket.put_object_from_file(target_key, local_path)
        
        # Optionally delete the local file upon successful upload
        if cleanup_source:
            os.remove(local_path)
            
        return True

Downloading Files

To retrieve objects from the cloud, the content is downloaded to a specified local path. If the path is omitted, the object key name is used as the filename.

    def download_file(self, remote_key, local_destination=None):
        if not self.check_object_exists(remote_key):
            raise FileNotFoundError(f"Object {remote_key} does not exist.")

        output_path = local_destination or remote_key
        self.bucket.get_object_to_file(remote_key, output_path)
        
        return True

Streaming Data Operations

For scenarios where files are not involved, raw bytes or file-like objects can be managed directly.

    def upload_data_stream(self, remote_key, data_bytes):
        """Upload raw bytes or a file-like object directly."""
        self.bucket.put_object(remote_key, data_bytes)
        return True

    def get_data_stream(self, remote_key):
        """Retrieve an object as a readable stream."""
        if not self.check_object_exists(remote_key):
            return None
        
        # Returns a file-like object allowing .read() operations
        return self.bucket.get_object(remote_key)

Object Deletion

Removing objects from the bucket is a straightforward operation using the object key.

    def delete_object(self, remote_key):
        self.bucket.delete_object(remote_key)
        return True

Generating Pre-Signed URLs

To grant temporary access to private objects without exposing credentials, a pre-signed URL can be generated. This URL expires after a defined time period.

    def generate_presigned_url(self, remote_key, expires_in_seconds=60, http_method='GET'):
        # Generate the signed URL
        signed_url = self.bucket.sign_url(http_method, remote_key, expires_in_seconds)
        
        # Remove internal endpoint markers if necessary for public access
        final_url = signed_url.replace("-internal", "")
        
        return final_url

Usage Examples

Below are examples demonstrating how to instantiate the manager and perform various operations.

# Configuration
oss_client = OSSManager(
    access_key_id='YOUR_ACCESS_KEY',
    access_key_secret='YOUR_SECRET',
    endpoint='https://oss-cn-hangzhou.aliyuncs.com',
    bucket_name='my-bucket'
)

# Upload a local file
oss_client.upload_file('/local/path/image.jpg', 'images/profile.jpg')

# Download a file
oss_client.download_file('images/profile.jpg', 'downloads/profile_backup.jpg')

# Upload via stream (e.g., text content)
content = b'Hello, OSS World'
oss_client.upload_data_stream('notes/hello.txt', content)

# Read stream content
stream = oss_client.get_data_stream('notes/hello.txt')
if stream:
    print(stream.read())

# Generate a temporary download link (valid for 5 minutes)
temp_url = oss_client.generate_presigned_url('images/profile.jpg', expires_in_seconds=300)
print(f"Shareable Link: {temp_url}")