如何将Python版Azure Function日志写入Azure Blob存储？

因为azure-storage-logging已废弃，直接使用官方的azure-storage-blob SDK实现日志追加是最可靠的方案。以下提供两种实现方式，均满足追加到同一Blob文件的需求：

方式一：自定义Logging Handler（自动捕获所有日志） #

通过自定义日志处理器，让所有logging输出自动追加到Blob，无需手动处理每条日志。

步骤1：安装依赖 #

在Function项目中安装SDK：

pip install azure-storage-blob

步骤2：完整代码示例 #

import logging
import azure.functions as func
from azure.storage.blob import AppendBlobClient
from datetime import datetime
import os

# 自定义日志处理器：将日志追加到Blob
class BlobAppendLogHandler(logging.Handler):
    def __init__(self, conn_str, container_name, blob_name):
        super().__init__()
        self.append_blob_client = AppendBlobClient.from_connection_string(conn_str, container_name, blob_name)
        # 若Blob不存在则自动创建
        try:
            self.append_blob_client.get_blob_properties()
        except:
            self.append_blob_client.create_append_blob()

    def emit(self, record):
        # 格式化日志：UTC时间 + 日志级别 + 消息
        log_content = f"{datetime.utcnow().isoformat()} [{record.levelname}] {record.getMessage()}\n"
        # 追加到Blob（自动处理编码）
        self.append_blob_client.append_block(log_content.encode('utf-8'))

app = func.FunctionApp()

# 初始化日志处理器（从环境变量读取配置，避免硬编码）
STORAGE_CONN_STR = os.environ["AZURE_STORAGE_CONNECTION_STRING"]
LOG_CONTAINER = "function-logs"
LOG_BLOB_NAME = "timer-function.log"

blob_log_handler = BlobAppendLogHandler(STORAGE_CONN_STR, LOG_CONTAINER, LOG_BLOB_NAME)
blob_log_handler.setLevel(logging.INFO)  # 只捕获INFO及以上级别日志
logging.getLogger().addHandler(blob_log_handler)

@app.schedule(schedule="*/30 * * * *", arg_name="myTimer", run_on_startup=True,
              use_monitor=False) 
def func_demo_timer_trigger(myTimer: func.TimerRequest) -> None:
    if myTimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function executed.')

方式二：手动追加指定日志（灵活控制内容） #

如果只需要将特定日志写入Blob，可在函数内直接调用SDK完成追加操作。

代码示例 #

import logging
import azure.functions as func
from azure.storage.blob import AppendBlobClient
from datetime import datetime
import os

app = func.FunctionApp()

@app.schedule(schedule="*/30 * * * *", arg_name="myTimer", run_on_startup=True,
              use_monitor=False) 
def func_demo_timer_trigger(myTimer: func.TimerRequest) -> None:
    # 初始化Blob客户端
    conn_str = os.environ["AZURE_STORAGE_CONNECTION_STRING"]
    append_blob = AppendBlobClient.from_connection_string(conn_str, "function-logs", "timer-function.log")
    
    # 确保Blob存在
    try:
        append_blob.get_blob_properties()
    except:
        append_blob.create_append_blob()

    # 收集并写入日志
    if myTimer.past_due:
        log_msg = f"{datetime.utcnow().isoformat()} [INFO] The timer is past due!\n"
        append_blob.append_block(log_msg.encode('utf-8'))
        logging.info('The timer is past due!')

    log_msg = f"{datetime.utcnow().isoformat()} [INFO] Python timer trigger function executed.\n"
    append_blob.append_block(log_msg.encode('utf-8'))
    logging.info('Python timer trigger function executed.')

关键配置与注意事项 #

1. 环境变量配置：将存储账户连接字符串AZURE_STORAGE_CONNECTION_STRING添加到Function App的「配置」中，避免硬编码敏感信息。
2. 权限设置：确保Function App拥有目标存储容器的「存储Blob数据参与者」权限（可通过Azure门户的IAM配置）。
3. 生产环境优化：
   • 若使用托管身份，可替换连接字符串为DefaultAzureCredential，无需存储密钥：

     from azure.identity import DefaultAzureCredential
     append_blob = AppendBlobClient(
         account_url="https://your-storage-account.blob.core.windows.net",
         container_name="function-logs",
         blob_name="timer-function.log",
         credential=DefaultAzureCredential()
     )
     

   • 添加异常捕获逻辑，避免Blob操作失败导致函数执行中断。
4. Blob特性：Append Blob支持原子追加操作，并发执行的函数不会出现日志乱序或覆盖问题。

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