本地Python开发Azure Functions时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)
问题排查及解决步骤
修复文件路径错误
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}"修正输出绑定用法
- 输出绑定无需指定
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存储
- 输出绑定无需指定
启用函数运行
local.settings.json中AzureWebJobs.my_function.Disabled设为true会禁用函数,修改为false:"AzureWebJobs.my_function.Disabled": "false"完整修正代码示例
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

