全部產品
Search
文件中心

Object Storage Service:Python快速入門

更新時間:Nov 30, 2024

本文介紹如何快速使用OSS Python SDK完成常見操作,如建立儲存空間(Bucket)、上傳檔案(Object)、下載檔案等。

前提條件

配置憑證

使用RAM使用者AccessKey配置環境變數。

Linux

  1. 在命令列介面執行以下命令來將環境變數設定追加到~/.bashrc 檔案中。

    echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.bashrc
    echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.bashrc
  2. 執行以下命令使變更生效。

    source ~/.bashrc
  3. 執行以下命令檢查環境變數是否生效。

    echo $OSS_ACCESS_KEY_ID
    echo $OSS_ACCESS_KEY_SECRET

macOS

  1. 在終端中執行以下命令,查看預設Shell類型。

    echo $SHELL
  2. 根據預設Shell類型進行操作。

    Zsh

    1. 執行以下命令來將環境變數設定追加到 ~/.zshrc 檔案中。

      echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.zshrc
      echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.zshrc
    2. 執行以下命令使變更生效。

      source ~/.zshrc
    3. 執行以下命令檢查環境變數是否生效。

      echo $OSS_ACCESS_KEY_ID
      echo $OSS_ACCESS_KEY_SECRET

    Bash

    1. 執行以下命令來將環境變數設定追加到 ~/.bash_profile 檔案中。

      echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.bash_profile
      echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.bash_profile
    2. 執行以下命令使變更生效。

      source ~/.bash_profile
    3. 執行以下命令檢查環境變數是否生效。

      echo $OSS_ACCESS_KEY_ID
      echo $OSS_ACCESS_KEY_SECRET

Windows

CMD

  1. 在CMD中運行以下命令。

    setx OSS_ACCESS_KEY_ID "YOUR_ACCESS_KEY_ID"
    setx OSS_ACCESS_KEY_SECRET "YOUR_ACCESS_KEY_SECRET"
  2. 運行以下命令,檢查環境變數是否生效。

    echo %OSS_ACCESS_KEY_ID%
    echo %OSS_ACCESS_KEY_SECRET%

PowerShell

  1. 在PowerShell中運行以下命令。

    [Environment]::SetEnvironmentVariable("OSS_ACCESS_KEY_ID", "YOUR_ACCESS_KEY_ID", [EnvironmentVariableTarget]::User)
    [Environment]::SetEnvironmentVariable("OSS_ACCESS_KEY_SECRET", "YOUR_ACCESS_KEY_SECRET", [EnvironmentVariableTarget]::User)
  2. 運行以下命令,檢查環境變數是否生效。

    [Environment]::GetEnvironmentVariable("OSS_ACCESS_KEY_ID", [EnvironmentVariableTarget]::User)
    [Environment]::GetEnvironmentVariable("OSS_ACCESS_KEY_SECRET", [EnvironmentVariableTarget]::User)

安裝SDK

  1. 已安裝合適版本的Python運行環境,通過以下命令查看Python版本。

    OSS Python SDK適用於Python 2.6、2.7、3.3、3.4、3.5、3.6、3.7、3.8及以上版本。

    python -version

    如果當前計算環境沒有安裝Python運行環境,請下載安裝Python

  2. 安裝OSS Python SDK。

    執行以下命令安裝SDK。

    pip install oss2

運行樣本

運行以下程式碼範例以體驗OSS的完整使用流程:建立一個Bucket、上傳檔案、下載檔案、列舉檔案以及刪除檔案和Bucket。

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

# 配置日誌
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 檢查環境變數是否已設定
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)

# 從環境變數中擷取訪問憑證
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# 設定Endpoint和Region
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
region = "cn-hangzhou"

def generate_unique_bucket_name():
    # 擷取目前時間戳
    timestamp = int(time.time())
    # 產生0到9999之間的隨機數
    random_number = random.randint(0, 9999)
    # 構建唯一的Bucket名稱
    bucket_name = f"demo-{timestamp}-{random_number}"
    return bucket_name

# 產生唯一的Bucket名稱
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}")

# 主流程
if __name__ == '__main__':
    # 1. 建立Bucket
    create_bucket(bucket)
    # 2. 上傳檔案
    upload_file(bucket, 'test-string-file', b'Hello OSS, this is a test string.')
    # 3. 下載檔案
    download_file(bucket, 'test-string-file')
    # 4. 列出Bucket中的對象
    list_objects(bucket)
    # 5. 刪除Bucket中的對象
    delete_objects(bucket)
    # 6. 刪除Bucket
    delete_bucket(bucket)