在Azure Functions中使用Python读写CSV文件的问题排查

我在Azure Functions中使用Python时遇到CSV文件读写问题，代码如下：

import datetime
import logging

import azure.functions as func
import csv
import traceback

def aprint(info):
    logging.info("Azure Print: "+str(info))
def read_csv_file(file_path):
        with open(file_path, 'r') as csvfile:
            aprint('Inside the file')
            reader = csv.reader(csvfile)
            for row in reader:
                # Process each row of data
                aprint(row)

def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
              tzinfo=datetime.timezone.utc).isoformat()
    logging.info("Program Has Started")  
    read_csv_file('general_DB.csv')
    if mytimer.past_due:
        logging.info('The timer is past due!')
        
        logging.info("Process passed successfully(thanks to timer)")

    logging.info('Python timer trigger function ran at %s', utc_timestamp)

运行时日志信息如下：

2023-05-26T21:59:14Z   [Information]   Executing 'Functions.TimerTrigger1' (Reason='This function was programmatically called via the host APIs.', Id=)
2023-05-26T21:59:14Z   [Verbose]   Sending invocation id:
2023-05-26T21:59:14Z   [Verbose]   Posting invocation id:
2023-05-26T21:59:14Z   [Information]   Program Has Started
2023-05-26T21:59:14Z   [Information]   Azure Print: Starting Read
2023-05-26T21:59:14Z   [Information]   Azure Print: Entered the function
2023-05-26T21:59:14Z   [Error]   Executed 'Functions.TimerTrigger1' (Failed, Id=, Duration=90ms)

已将CSV文件与.py文件放在云端同一目录下，排查后确定问题出在文件打开环节，寻求解决方案。

解决方案 #

• 修正文件路径获取逻辑
  Azure Functions运行时的当前工作目录并非函数代码所在目录，直接使用文件名会导致找不到文件。需通过脚本路径拼接正确的CSV文件路径：

import os
# ... 保留原有导入代码

def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
              tzinfo=datetime.timezone.utc).isoformat()
    logging.info("Program Has Started")  
    # 获取当前脚本所在目录
    script_dir = os.path.dirname(os.path.abspath(__file__))
    # 拼接CSV文件完整路径
    csv_full_path = os.path.join(script_dir, 'general_DB.csv')
    read_csv_file(csv_full_path)
    # ... 保留后续代码

• 添加异常捕获与日志输出
  在文件操作部分增加异常捕获，打印详细错误信息，便于定位具体问题：

def read_csv_file(file_path):
    try:
        # 指定编码避免乱码或读取失败
        with open(file_path, 'r', encoding='utf-8') as csvfile:
            aprint('Inside the file')
            reader = csv.reader(csvfile)
            for row in reader:
                aprint(row)
    except Exception as e:
        aprint(f"文件操作错误: {str(e)}")
        # 打印完整堆栈追踪信息
        aprint(traceback.format_exc())

• 确认文件部署状态
  登录Azure门户，进入函数应用→高级工具(Kudu)→调试控制台→文件管理器，导航到site/wwwroot/你的函数名称目录，确认general_DB.csv文件存在且内容正确。若文件缺失，重新部署函数代码及CSV文件。

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