Azure函数报错：无法从azure.storage.blob导入AppendBlobService

问题描述 #

• 运行环境：HTTP触发的Python Azure函数，代码分为__init__.py入口文件和同目录下的test.py业务逻辑文件
• 异常表现：代码在PyCharm、Databricks环境可正常运行，通过VS Code本地运行Azure函数时抛出导入错误：

  Exception: ImportError: cannot import name 'AppendBlobService' from 'azure.storage.blob'
  

• 已尝试操作：执行pip install azure-storage --upgrade、pip install azure-storage-blob命令升级依赖，问题未解决
• 相关代码片段：
  入口文件__init__.py：

  import logging
  import azure.functions as func
  from . import test
  
  def main(req: func.HttpRequest) -> func.HttpResponse:
      logging.info('Python HTTP trigger function processed a request.')
  
      name = req.params.get('name')
      strng = req.params.get('strng')
      if not name:
          try:
              req_body = req.get_json()
          except ValueError:
              pass
          else:
              name = req_body.get('name')
  
      if name:
          return func.HttpResponse(f"Hello, {name}. This HTTP triggered function executed successfully.sum = {test.testfunc(strng)}")
      else:
          return func.HttpResponse(
               "This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
               status_code=200
          )
  

  业务文件test.py：

  import json
  import pandas as pd
  from pandas import DataFrame
  from azure.storage.blob import AppendBlobService
  from datetime import datetime
  
  
  def testfunc(strng):
      # return strng
      API = json.loads(strng)
  
      test = pd.json_normalize(parse, record_path='Vol', meta=['studyDate'])
      df = pd.DataFrame(test)
      df["x"] = df["Vol"] * 2
      df["y"] = df["Vol"] * 50
      df1 = df[['Date', 'x', 'y']]
      df2 = df1.to_json(orient='records')
      append_blob_service = AppendBlobService(account_name='actname',
                                           account_key='key')
      date = datetime.now()
      blobname = f"test_cal{date}.json"
      append_blob_service.create_blob('container', blobname, if_none_match="*")
      append_blob_service.append_blob_from_text('container', blobname, text=df2)
      return df2
  

根因说明 #

AppendBlobService是v2.x旧版azure-storage-blob SDK中的类，在v12+的新版SDK中已经被移除，相关操作整合到了BlobServiceClient中。
之前执行的安装命令默认拉取最新版SDK，因此找不到对应类；PyCharm和Databricks环境中安装的是旧版SDK，因此可以正常运行。另外azure-storage是早已停止维护的旧元包，和新版blob SDK存在依赖冲突，安装后会进一步加剧导入异常。
另外业务代码中存在一个隐藏bug：pd.json_normalize传入的parse变量未定义，正常应该传入前面解析得到的API变量，修复导入问题后需要同步修正这个点，否则会触发NameError。

修复方案 #

方案1：固定旧版SDK版本（快速兼容现有代码） #

如果不想修改现有Blob操作逻辑，可以卸载冲突依赖后安装指定的旧版本SDK：

1. 执行以下命令清理并重装依赖：

   pip uninstall -y azure-storage azure-storage-blob
   pip install azure-storage-blob==2.1.0
   

2. 在Azure函数项目根目录的requirements.txt文件中添加固定版本声明，避免后续部署或重装依赖时拉取错版本：

   azure-storage-blob==2.1.0
   pandas
   azure-functions
   

3. 修正test.py中pd.json_normalize的参数，把parse替换为API。

方案2：使用新版v12+ SDK（长期维护推荐） #

旧版v2.x SDK已经停止功能更新，建议迁移到新版SDK实现Blob写入，操作更稳定：

1. 执行以下命令安装新版依赖：

   pip uninstall -y azure-storage
   pip install azure-storage-blob>=12.0.0 pandas azure-functions
   

2. 修改test.py中的Blob操作逻辑，替换导入语句和写入代码，修正后的完整test.py参考如下：

   import json
   import pandas as pd
   import os
   from azure.storage.blob import BlobServiceClient
   from datetime import datetime
   
   
   def testfunc(strng):
       API = json.loads(strng)
       # 修正原来未定义的parse变量，替换为解析得到的API
       test = pd.json_normalize(API, record_path='Vol', meta=['studyDate'])
       df = pd.DataFrame(test)
       df["x"] = df["Vol"] * 2
       df["y"] = df["Vol"] * 50
       df1 = df[['Date', 'x', 'y']]
       df2 = df1.to_json(orient='records')
       
       # 新版SDK初始化Blob客户端，生产环境建议把连接串/密钥存在函数应用配置中，通过os.getenv读取，不要硬编码
       connect_str = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
       blob_service_client = BlobServiceClient.from_connection_string(connect_str)
       container_client = blob_service_client.get_container_client("container")
       
       date = datetime.now()
       blobname = f"test_cal{date}.json"
       blob_client = container_client.get_blob_client(blobname)
       
       # 创建追加Blob，已存在则跳过
       try:
           blob_client.create_append_blob()
       except Exception:
           pass
       # 追加文本内容
       blob_client.append_block(df2)
       
       return df2
   

注意：不要在代码中硬编码存储账号密钥，生产环境请将存储凭据配置在Azure函数的应用设置中，通过环境变量读取，避免凭据泄露。

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