DataWorks：節點調度配置

方式一：讀取環境變數（推薦 Shell, Python）

系統會自動化佈建以下三個環境變數，您可以直接在代碼中讀取。

• LINKED_ROLE_ACCESS_KEY_ID：臨時存取金鑰 ID。

• LINKED_ROLE_ACCESS_KEY_SECRET：臨時存取金鑰 Secret。

• LINKED_ROLE_SECURITY_TOKEN：臨時安全性權杖。

程式碼範例 (Python)：

import os
import oss2

# 1. 從環境變數擷取臨時憑證
access_key_id = os.environ.get('LINKED_ROLE_ACCESS_KEY_ID')
access_key_secret = os.environ.get('LINKED_ROLE_ACCESS_KEY_SECRET')
security_token = os.environ.get('LINKED_ROLE_SECURITY_TOKEN')

# 檢查是否成功擷取
if not all([access_key_id, access_key_secret, security_token]):
    raise Exception("Failed to get linked role credentials from environment variables.")

# 2. 使用臨時憑證初始化 OSS 用戶端
# 假設您已為角色授予了訪問 'your-bucket-name' 的許可權
auth = oss2.StsAuth(access_key_id, access_key_secret, security_token)
bucket = oss2.Bucket(auth, 'http://oss-<regionID>-internal.aliyuncs.com', 'your-bucket-name')

# 3. 使用用戶端訪問 OSS 資源
try:
    # 列出 bucket 中的檔案
    for obj in oss2.ObjectIterator(bucket):
        print('object name: ' + obj.key)
    print("Successfully accessed OSS with linked role.")
except oss2.exceptions.OssError as e:
    print(f"Error accessing OSS: {e}")

程式碼範例 (Shell)：

#!/bin/bash
access_key_id=${LINKED_ROLE_ACCESS_KEY_ID}
access_key_secret=${LINKED_ROLE_ACCESS_KEY_SECRET}
security_token=${LINKED_ROLE_SECURITY_TOKEN}

# 訪問OSS，需替換regionID、bucket_name和file_name為真實資訊
echo "ID："$access_key_id
echo "token:"$security_token
ls -al /home/admin/usertools/tools/

# 樣本為通過ossutil從OSS指定路徑下載檔案到本地test_dw.py，並輸出檔案內容。
/home/admin/usertools/tools/ossutil64 cp --access-key-id $access_key_id --access-key-secret $access_key_secret --sts-token $security_token --endpoint http://oss-<regionID>-internal.aliyuncs.com oss://<bucket_name>/<file_name> test_dw.py
echo "************************ 擷取成功 ************************，列印結果"
cat test_dw.py

方式二：使用 Credentials Client (推薦 Python)

程式碼範例 (Python)：

from alibabacloud_credentials.client import Client as CredentialClient
import oss2

# 1. 使用 SDK 自動擷取憑證
# 它會自動尋找環境變數中的 LINKED_ROLE_* 等憑證資訊
cred_client = CredentialClient()
credential = cred_client.get_credential()

access_key_id = credential.get_access_key_id()
access_key_secret = credential.get_access_key_secret()
security_token = credential.get_security_token()

if not all([access_key_id, access_key_secret, security_token]):
    raise Exception("Failed to get linked role credentials via SDK.")

# 2. 使用憑證初始化 OSS 用戶端
auth = oss2.StsAuth(access_key_id, access_key_secret, security_token)
bucket = oss2.Bucket(auth, 'http://oss-cn-hangzhou.aliyuncs.com', 'your-bucket-name')

# 3. 訪問 OSS
print("Listing objects in bucket...")
for obj in oss2.ObjectIterator(bucket):
    print(' - ' + obj.key)
print("Successfully accessed OSS with linked role via SDK.")