如何用Python将日志写入Azure容器Blob？basicConfig报错求助

Python标准库的logging.basicConfig中filename参数仅支持本地文件系统路径，无法直接识别Azure Blob存储的HTTP URL。你传入的模拟器URL被当作本地文件路径处理，Windows系统文件名不允许包含:、/这类特殊字符，因此触发OSError: [Errno 22] Invalid argument错误。

通过Azure Storage Blob SDK实现自定义日志Handler，直接将日志写入指定Blob容器的文件夹中，步骤如下：

1. 安装依赖 #

执行以下命令安装Azure Blob存储SDK：

pip install azure-storage-blob

2. 完整实现代码 #

import logging
from datetime import date
from azure.storage.blob import BlobServiceClient
from logging import Handler, LogRecord

class BlobLoggingHandler(Handler):
    def __init__(self, connection_string, container_name, blob_prefix):
        super().__init__()
        self.blob_service_client = BlobServiceClient.from_connection_string(connection_string)
        self.container_name = container_name
        self.blob_prefix = blob_prefix
        # 确保目标容器存在
        self.blob_service_client.create_container(container_name, exist_ok=True)

    def emit(self, record: LogRecord):
        # 格式化单条日志
        log_entry = self.format(record) + "\n"
        today = date.today().strftime("%Y-%m-%d")
        blob_name = f"{self.blob_prefix}/{today}_logs.log"
        
        # 获取Blob客户端
        blob_client = self.blob_service_client.get_blob_client(container=self.container_name, blob=blob_name)
        
        # 实现追加写入：读取现有内容（如果存在），拼接新日志后覆盖上传
        try:
            existing_content = blob_client.download_blob().readall().decode("utf-8")
            new_content = existing_content + log_entry
        except:
            # Blob不存在时直接写入新内容
            new_content = log_entry
        
        blob_client.upload_blob(new_content, overwrite=True)

# 配置日志系统
def setup_logging():
    # 替换为你的Azure存储连接字符串（模拟器或生产环境）
    # 本地模拟器连接字符串示例
    connection_string = "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;"
    container_name = "cloudops-resources"
    blob_prefix = "logs"

    # 创建并配置自定义Handler
    blob_handler = BlobLoggingHandler(connection_string, container_name, blob_prefix)
    formatter = logging.Formatter("%(asctime)s,%(msecs)d %(name)s %(message)s", datefmt="%H:%M:%S")
    blob_handler.setFormatter(formatter)
    blob_handler.setLevel(logging.INFO)

    # 绑定到根日志器
    root_logger = logging.getLogger()
    root_logger.setLevel(logging.INFO)
    root_logger.addHandler(blob_handler)
    # 可选：添加控制台输出Handler
    # console_handler = logging.StreamHandler()
    # console_handler.setFormatter(formatter)
    # root_logger.addHandler(console_handler)

# 初始化日志配置
setup_logging()

# 测试日志写入
logger = logging.getLogger(__name__)
logger.info("测试日志条目1")
logger.info("测试日志条目2")

代码说明 #

• BlobLoggingHandler：继承自logging.Handler，重写emit方法实现日志写入Blob的逻辑，支持追加模式（兼容已有日志文件）。
• 连接字符串：根据环境替换，本地模拟器使用固定连接字符串，生产环境需从Azure Portal获取真实连接字符串。
• 容器与路径：container_name为目标容器名，blob_prefix对应容器内的日志文件夹路径。

如果不需要实时写入Blob，也可以先将日志写入本地文件，再手动/定期上传到Azure Blob：

import logging
from datetime import date
from azure.storage.blob import BlobServiceClient

# 配置本地日志
today = date.today()
local_log_path = f"{today}_logs.log"
logging.basicConfig(
    filename=local_log_path,
    filemode='a',
    format='%(asctime)s,%(msecs)d %(name)s %(message)s',
    datefmt='%H:%M:%S',
    level=logging.INFO,
    force=True
)

# 生成测试日志
logger = logging.getLogger(__name__)
logger.info("本地测试日志")

# 上传到Azure Blob
connection_string = "你的连接字符串"
container_name = "cloudops-resources"
blob_name = f"logs/{today}_logs.log"

blob_service_client = BlobServiceClient.from_connection_string(connection_string)
blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)

with open(local_log_path, "rb") as data:
    blob_client.upload_blob(data, overwrite=True)

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