如何在Azure Function中用Python访问Blob存储并读取Excel文件

最简访问方式 #

最直接的实现方式是使用Azure Storage Blob SDK（azure-storage-blob），配合两种认证方案：

• 存储账户连接字符串：配置步骤最少，直接通过连接字符串访问Blob，适合快速测试或小型场景。
• 托管标识（Managed Identity）：无需存储敏感连接字符串，通过Azure身份系统授权，安全等级更高，适合生产环境。

前置准备 #

1. 在Function的requirements.txt中添加依赖包：

azure-functions
azure-storage-blob>=12.0.0
pandas>=2.0.0
openpyxl>=3.0.0  # 读取.xlsx文件需安装；读取.xls则替换为xlrd>=2.0.0

2. 配置认证信息：
   • 连接字符串方式：在Azure Function应用设置中新增AZURE_STORAGE_CONNECTION_STRING，值为Blob存储账户的连接字符串。
   • 托管标识方式：为Function的系统托管标识分配「存储Blob数据读取者」权限到目标存储账户或容器。

Python读取代码示例 #

示例1：HTTP触发，通过连接字符串读取指定Excel文件 #

import azure.functions as func
from azure.storage.blob import BlobServiceClient
import pandas as pd
import io
import os

def main(req: func.HttpRequest) -> func.HttpResponse:
    # 从应用设置获取连接字符串
    conn_str = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
    if not conn_str:
        return func.HttpResponse("未配置存储连接字符串", status_code=400)

    # 初始化Blob服务客户端
    blob_service_client = BlobServiceClient.from_connection_string(conn_str)
    
    # 替换为你的容器名和Excel文件路径
    container_name = "your-container-name"
    blob_file_path = "folder/subfolder/test.xlsx"

    try:
        # 获取Blob客户端并下载内容
        blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_file_path)
        blob_bytes = blob_client.download_blob().readall()
        
        # 读取Excel为DataFrame
        df = pd.read_excel(io.BytesIO(blob_bytes), engine="openpyxl")
        
        # 此处可添加数据处理逻辑，示例返回前5行数据
        return func.HttpResponse(f"读取成功，数据预览：\n{df.head().to_string()}", status_code=200)
    except Exception as e:
        return func.HttpResponse(f"读取失败：{str(e)}", status_code=500)

示例2：Blob触发，自动读取新上传的Excel文件 #

如果需要在Excel文件上传到Blob时自动触发读取，用Blob触发的Function更简便：

import azure.functions as func
import pandas as pd
import io

def main(myblob: func.InputStream) -> None:
    # 直接读取触发的Blob内容
    blob_bytes = myblob.read()
    df = pd.read_excel(io.BytesIO(blob_bytes), engine="openpyxl")
    
    # 示例：打印数据行数，可替换为你的业务逻辑
    print(f"成功读取文件 {myblob.name}，数据共 {len(df)} 行")

示例3：托管标识认证方式（生产环境推荐） #

替换示例1中的Blob服务客户端初始化代码即可：

from azure.identity import DefaultAzureCredential

# 替换为你的存储账户URL，格式为 https://{存储账户名}.blob.core.windows.net
account_url = "https://your-storage-account.blob.core.windows.net"
blob_service_client = BlobServiceClient(account_url=account_url, credential=DefaultAzureCredential())

内容的提问来源于stack exchange，提问作者otaku