本文介紹如何快速使用OSS Python SDK完成常見操作,如建立儲存空間(Bucket)、上傳檔案(Object)、下載檔案等。
注意事項
關於OSS支援的Region與Endpoint的對應關係,請參見OSS地區和訪問網域名稱。
本文以從環境變數讀取存取憑證為例。更多配置訪問憑證的樣本,請參見配置訪問憑證(Python SDK V1)。
本文通過預設樣本對Python SDK進行初始化。更多情境樣本,請參見初始化(Python SDK V1)。
前提條件
配置憑證
使用RAM使用者AccessKey配置環境變數。
Linux
在命令列介面執行以下命令來將環境變數設定追加到
~/.bashrc檔案中。echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.bashrc echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.bashrc執行以下命令使變更生效。
source ~/.bashrc執行以下命令檢查環境變數是否生效。
echo $OSS_ACCESS_KEY_ID echo $OSS_ACCESS_KEY_SECRET
macOS
在終端中執行以下命令,查看預設Shell類型。
echo $SHELL根據預設Shell類型進行操作。
Zsh
執行以下命令來將環境變數設定追加到
~/.zshrc檔案中。echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.zshrc echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.zshrc執行以下命令使變更生效。
source ~/.zshrc執行以下命令檢查環境變數是否生效。
echo $OSS_ACCESS_KEY_ID echo $OSS_ACCESS_KEY_SECRET
Bash
執行以下命令來將環境變數設定追加到
~/.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執行以下命令使變更生效。
source ~/.bash_profile執行以下命令檢查環境變數是否生效。
echo $OSS_ACCESS_KEY_ID echo $OSS_ACCESS_KEY_SECRET
Windows
CMD
在CMD中運行以下命令。
setx OSS_ACCESS_KEY_ID "YOUR_ACCESS_KEY_ID" setx OSS_ACCESS_KEY_SECRET "YOUR_ACCESS_KEY_SECRET"運行以下命令,檢查環境變數是否生效。
echo %OSS_ACCESS_KEY_ID% echo %OSS_ACCESS_KEY_SECRET%
PowerShell
在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)運行以下命令,檢查環境變數是否生效。
[Environment]::GetEnvironmentVariable("OSS_ACCESS_KEY_ID", [EnvironmentVariableTarget]::User) [Environment]::GetEnvironmentVariable("OSS_ACCESS_KEY_SECRET", [EnvironmentVariableTarget]::User)
參考上述方式修改系統內容變數後,請重啟或重新整理您的編譯運行環境,包括IDE、命令列介面、其他傳統型應用程式及後台服務,以確保最新的系統內容變數成功載入。
安裝SDK
已安裝合適版本的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。
安裝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)

