You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

本地Python开发Azure Functions时Blob输出绑定不生成文件求助

Azure Functions Python Blob输出绑定失效及文件保存错误排查

本地环境(Azure存储模拟器+Azure Storage Explorer)开发Python Azure Functions,需求是Blob容器cloudops-resources下的inputs文件夹上传Excel时触发函数,处理后将结果写入同容器的outputs文件夹。目前输入触发正常,但输出绑定未生效,修改代码后出现文件找不到错误。

配置文件

1. function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "inputblob",
      "type": "blobTrigger",
      "direction": "in",
      "path": "cloudops-resources/inputs/{name}",
      "connection": "AzureWebJobsStorage"
    },
    {
    "name": "outputblob",
    "type": "blob",
    "direction": "out",
    "path": "cloudops-resources/outputs/{name}",
    "connection": "AzureWebJobsStorage"
    }
  ],
  "disabled": false
}

2. local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "AzureWebJobs.my_function.Disabled": "true"
  }
}

代码及错误

初始未生效代码

def main(inputblob: func.InputStream, outputblob: func.Out[func.InputStream]):
    logging.info(f"Python blob trigger function processed blob \n"
                 f"Name: {inputblob.name}\n"
                 f"Blob Size: {inputblob.length} bytes")

    # some code here

    # Reading the input xlsx file from customer
    df0 = pd.read_excel(inputblob)

    # Opening a new workbook and a worksheet
    wb = Workbook()
    sheet = wb.active

    # some code here

    # Saving the file
    today = date.today()
    date_today = today.strftime("%m-%d-%y")
    outputblob = "{}_COMPLETE_{}.xlsx".format(date_today,inputblob)
    wb.save(outputblob)
    return func.Out[func.InputStream](
         body=outputblob,
         status_code=200
    )

修改后报错代码

def main(inputblob: func.InputStream, outputblob: func.Out[func.InputStream]):
    logging.info(f"Python blob trigger function processed blob \n"
                 f"Name: {inputblob.name}\n"
                 f"file: {inputblob}\n"
                 f"Blob Size: {inputblob.length} bytes")

    # some code here

    df0 = pd.read_excel(inputblob.read())
    logging.info(f"{df0}")

    # Opening a new workbook and a worksheet
    wb = Workbook()
    sheet = wb.active

    # some code here

    # Saving the file
    today = date.today()
    date_today = today.strftime("%m-%d-%y")
    outputfile = "{}_COMPLETE_{}".format(date_today,inputblob.name)
    wb.save(outputfile)

    outputblob.set(wb)
    logging.info(f"Name: {outputblob.name}\n"
                 f"file: {outputblob}\n"
                 f"Blob Size: {outputblob.length} bytes")

    return()

错误信息

System.Private.CoreLib: Exception while executing function: Functions.BlobTrigger. System.Private.CoreLib: Result: Failure
Exception: FileNotFoundError: [Errno 2] No such file or directory: '03-29-23_COMPLETE_cloudops-resources/inputs/Input_file123.xlsx'
Stack: File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.9/WINDOWS/X64\azure_functions_worker\dispatcher.py", line 452, in _handle__invocation_request
call_result = await self._loop.run_in_executor(
File "C:\Users\Asu\AppData\Local\Programs\Python\Python39\lib\concurrent\futures\thread.py", line 58, in run
result = self.fn(*self.args, **self.kwargs)
File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.9/WINDOWS/X64\azure_functions_worker\dispatcher.py", line 718, in run_sync_func
return ExtensionManager.get_sync_invocation_wrapper(context,
File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.9/WINDOWS/X64\azure_functions_worker\extension.py", line 215, in raw_invocation_wrapper
result = function(**args)
File "D:\Service Projects\Customers\InvoiceCompare\BlobTrigger___init
.py", line 360, in main wb.save(outputfile)
File "D:\Service Projects\Customers\InvoiceCompare\env\lib\site-packages\openpyxl\workbook\workbook.py", line 386, in save
save_workbook(self, filename)
File "D:\Service Projects-1\Customers\InvoiceCompare\env\lib\site-packages\openpyxl\writer\excel.py", line 291, in save_workbook
archive = ZipFile(filename, 'w', ZIP_DEFLATED, allowZip64=True)
File "C:\Users\Asu\AppData\Local\Programs\Python\Python39\lib\zipfile.py", line 1246, in init
self.fp = io.open(file, filemode)

问题排查及解决步骤

  1. 修复文件路径错误
    inputblob.name返回Blob的完整路径(如cloudops-resources/inputs/Input_file123.xlsx),直接拼接会生成不存在的嵌套路径。需提取纯文件名:

    from os.path import basename
    # ...
    original_filename = basename(inputblob.name)
    output_filename = f"{date_today}_COMPLETE_{original_filename}"
    
  2. 修正输出绑定用法

    • 输出绑定无需指定func.InputStream,直接使用func.Out[bytes]接收字节流
    • 避免保存到本地文件,将Workbook写入内存字节流后传入输出绑定:
      from io import BytesIO
      # ...
      output_stream = BytesIO()
      wb.save(output_stream)
      output_stream.seek(0)  # 重置流指针到开头
      outputblob.set(output_stream.getvalue())  # 写入Blob存储
      
  3. 启用函数运行
    local.settings.json中AzureWebJobs.my_function.Disabled设为true会禁用函数,修改为false:

    "AzureWebJobs.my_function.Disabled": "false"
    
  4. 完整修正代码示例

    import logging
    import os
    from datetime import date
    from io import BytesIO
    import azure.functions as func
    import pandas as pd
    from openpyxl import Workbook
    
    def main(inputblob: func.InputStream, outputblob: func.Out[bytes]):
        logging.info(f"Python blob trigger function processed blob \n"
                     f"Name: {inputblob.name}\n"
                     f"Blob Size: {inputblob.length} bytes")
    
        # 读取输入Excel
        df0 = pd.read_excel(inputblob.read())
        logging.info(f"Read data: {df0}")
    
        # 自定义处理逻辑
        wb = Workbook()
        sheet = wb.active
        sheet.append(["Processed", "Data"])
        for idx, row in df0.iterrows():
            sheet.append(row.tolist())
    
        # 生成输出文件名
        today = date.today()
        date_today = today.strftime("%m-%d-%y")
        original_filename = os.path.basename(inputblob.name)
        output_filename = f"{date_today}_COMPLETE_{original_filename}"
        logging.info(f"Output filename: {output_filename}")
    
        # 写入内存流并输出到Blob
        output_stream = BytesIO()
        wb.save(output_stream)
        output_stream.seek(0)
        outputblob.set(output_stream.getvalue())
        logging.info(f"Successfully wrote output blob")
    

内容的提问来源于stack exchange,提问作者Asu

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.26 23:12:29