Object Storage Service:OSS SDK for Pythonの使用を開始する

前提条件

• Alibaba Cloudアカウントが作成されました。

• Object Storage Service (OSS) が有効化されています。

• OSSを管理する権限を持つResource Access Management (RAM) ユーザー用にAccessKeyペアが作成されます。

  Resource Orchestration Service (ROS) スクリプトを使用して、OSSを管理する権限を持つRAMユーザーのAccessKeyペアをすばやく作成します。

  [スタックの作成] ウィザードの [パラメータの設定] ステップで、[Alibaba Cloud ROSがRAMリソースを作成できることを確認] を選択し、[セキュリティ確認] セクションの [作成] をクリックします。

  

  AccessKeyペアが作成されたら、[出力] タブでAccessKeyペアをコピーします。

  

AccessKeyペアの設定

RAMユーザーのAccessKeyペアを使用して、環境変数を設定します。

OSS SDK for Pythonのインストール

1. Pythonランタイム環境がインストールされている場合、次のコマンドを実行してPythonランタイム環境のバージョンを表示します。

   Python用OSS SDKには、Python 2.6、2.7、3.3、3.4、3.5、3.6、3.7、3.8、またはそれ以降が必要です。

   python -version

   Pythonランタイム環境がインストールされていない場合は、Pythonランタイム環境をダウンロードしてインストールします。

2. OSS SDK for Pythonをインストールします。

   次のコマンドを実行して、OSS SDK for Pythonをインストールします。

   pip install oss2

例

次のサンプルコードを実行してバケットを作成および削除し、オブジェクトをアップロード、ダウンロード、リスト、および削除して、OSS SDK for Pythonを使用して一般的な操作を実行する方法を学習します。

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
from itertools import islice
import os
import logging
import time
import random

# Configure logs.
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Check whether the environment variables are configured.
required_env_vars = ['OSS_ACCESS_KEY_ID', 'OSS_ACCESS_KEY_SECRET']
for var in required_env_vars:
    if var not in os.environ:
        logging.error(f"Environment variable {var} is not set.")
        exit(1)

# Obtain access credentials from environment variables.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Specify the endpoint and region.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
region = "cn-hangzhou"

def generate_unique_bucket_name():
    # Query the current timestamp.
    timestamp = int(time.time())
    # Generate a random number from 0 to 9999.
    random_number = random.randint(0, 9999)
    # Specify the name of the bucket. The name must be globally unique.
    bucket_name = f"demo-{timestamp}-{random_number}"
    return bucket_name

# Generate a unique bucket name.
bucket_name = generate_unique_bucket_name()
bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)

def create_bucket(bucket):
    try:
        bucket.create_bucket(oss2.models.BUCKET_ACL_PRIVATE)
        logging.info("Bucket created successfully")
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to create bucket: {e}")

def upload_file(bucket, object_name, data):
    try:
        result = bucket.put_object(object_name, data)
        logging.info(f"File uploaded successfully, status code: {result.status}")
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to upload file: {e}")

def download_file(bucket, object_name):
    try:
        file_obj = bucket.get_object(object_name)
        content = file_obj.read().decode('utf-8')
        logging.info("File content:")
        logging.info(content)
        return content
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to download file: {e}")

def list_objects(bucket):
    try:
        objects = list(islice(oss2.ObjectIterator(bucket), 10))
        for obj in objects:
            logging.info(obj.key)
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to list objects: {e}")

def delete_objects(bucket):
    try:
        objects = list(islice(oss2.ObjectIterator(bucket), 100))
        if objects:
            for obj in objects:
                bucket.delete_object(obj.key)
                logging.info(f"Deleted object: {obj.key}")
        else:
            logging.info("No objects to delete")
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to delete objects: {e}")

def delete_bucket(bucket):
    try:
        bucket.delete_bucket()
        logging.info("Bucket deleted successfully")
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to delete bucket: {e}")

# The following sample code provides an example of the main process of using OSS SDK for Python to perform common operations.
if __name__ == '__main__':
    # 1: Create a bucket.
    create_bucket(bucket)
    # 2. Upload a local file to the bucket.
    upload_file(bucket, 'test-string-file', b'Hello OSS, this is a test string.')
    # 3. Download an object from the bucket.
    download_file(bucket, 'test-string-file')
    # 4. List objects in the bucket.
    list_objects(bucket)
    # 5. Delete an object from the bucket.
    delete_objects(bucket)
    # 6. Delete a bucket.
    delete_bucket(bucket)