我有一个Python工具想要与Azure SDK交互(列出和上传blob)。通常我会使用类似的东西:
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient
creds = DefaultAzureCredential()
client = BlobServiceClient(..., credential=creds)
考虑到我依赖通过 CLI 中当前登录的用户进行身份验证,我想在 GitHub Action 工作流程中运行此脚本。
问题是 GitHub 操作中的身份验证方法必须通过联合凭据。
我有一个配置了联合身份的服务主体,我可以使用
azure/login
操作进行身份验证和登录。问题是,在必须运行 Python 脚本的步骤中,上一步中登录的 CLI 的上下文似乎没有保留,因为我看到以下错误:
azure.core.exceptions.HttpResponseError: This request is not authorized to perform this operation using this permission.
我的工作流程如下(简化):
jobs:
execute:
runs-on: ubuntu-latest
steps:
# checkout and other stuff
# ...
- name: AZ Login
uses: Azure/login@v1
with:
client-id: ${{ vars.AZURE_CLIENT_ID }}
subscription-id: ${{ vars.AZURE_SUBSCRIPTION_ID }}
tenant-id: ${{ vars.AZURE_TENANT_ID }}
# setup python and install packages
# ...
- name: Run script
run: |
.venv/bin/python script.py
问题是:如何将凭证从
azure/login
步骤传播到 python 运行时环境中?
This request is not authorized to perform this operation using this permission.
如果服务主体没有读取或执行操作的权限,您将收到上述错误。要解决此错误,请将
storage data contributor
角色添加到 Blob 存储帐户的服务原则中,如下所示:
步骤1:进入存储账户的IAM,点击Add
,选择Add role assignment
,如下图:
Step2:搜索
storage data contributor
角色并选择,如下图:
Step3:选择服务原则如下:
成功分配角色后,您将能够使用以下代码读取文件或列出容器中的文件:
import io
from azure.storage.blob import BlobServiceClient
import pandas as pd
from azure.identity import ClientSecretCredential
account_url = "https://<storageAccountName>.blob.core.windows.net"
container_name = "<containerNmae>"
file_path = "<filePath>"
TENANT_ID = "<tenantId>"
CLIENT_ID = "clientId"
CLIENT_SECRET = "clientSecret"
credentials = ClientSecretCredential(TENANT_ID, CLIENT_ID, CLIENT_SECRET)
blob_service_client = BlobServiceClient(account_url=account_url, credential=credentials)
container_client = blob_service_client.get_container_client(container_name)
blobs_list = container_client.list_blobs()
for blob in blobs_list:
print(blob.name)
blob_client = container_client.get_blob_client(file_path)
blob_data = blob_client.download_blob()
csv_data = blob_data.content_as_text()
df = pd.read_csv(io.StringIO(csv_data))
print("Sample.CSV data:")
print(df)