Python Azure Timer Function每日运行，如何实现每月任务？全局变量存疑

你的担忧是对的——Azure Timer Function的全局变量无法跨触发持久化，函数冷启动时会重新初始化变量，导致你原有的逻辑失效。下面是两种可靠的实现方式：

方案一：直接基于日期判断（无需额外存储） #

如果每月任务是固定在某一天执行（比如每月1号、最后一天），直接通过当前日期判断即可，不用维护下次执行时间。

示例代码：

from datetime import datetime
import calendar

def main(timer):
    current_date = datetime.now().date()
    
    # 判断是否是当月第一天
    if current_date.day == 1:
        # 执行每月特定任务
        print("执行月度任务")
    else:
        # 执行每日常规任务
        print("执行每日任务")

    # 或者判断是否是当月最后一天
    # _, last_day = calendar.monthrange(current_date.year, current_date.month)
    # if current_date.day == last_day:
    #     # 执行月度任务

方案二：用Azure Storage持久化执行标记（适合非固定日期的场景） #

如果月度任务需要按间隔（比如每30天）执行，而非固定日期，需要把下次执行时间存在外部存储（比如Azure Blob Storage），避免内存变量丢失。

步骤1：安装依赖 #

在函数项目的requirements.txt中添加：

azure-storage-blob>=12.0.0

步骤2：实现代码 #

from datetime import datetime, timedelta
from azure.storage.blob import BlobServiceClient
import os

# 从环境变量获取存储连接字符串
CONNECT_STR = os.environ["AzureWebJobsStorage"]
BLOB_CONTAINER = "job-markers"
BLOB_NAME = "next_monthly_job.txt"

def get_next_monthly_time():
    """从Blob存储读取下次月度任务执行时间"""
    blob_service_client = BlobServiceClient.from_connection_string(CONNECT_STR)
    try:
        blob_client = blob_service_client.get_blob_client(container=BLOB_CONTAINER, blob=BLOB_NAME)
        if blob_client.exists():
            time_str = blob_client.download_blob().readall().decode("utf-8")
            return datetime.fromisoformat(time_str)
        else:
            # 首次运行，设置30天后的时间
            next_time = datetime.now() + timedelta(days=30)
            blob_client.upload_blob(next_time.isoformat())
            return next_time
    except Exception as e:
        print(f"读取存储失败: {e}")
        # 异常时默认不执行月度任务
        return datetime.max

def update_next_monthly_time():
    """更新Blob存储中的下次月度任务执行时间"""
    blob_service_client = BlobServiceClient.from_connection_string(CONNECT_STR)
    blob_client = blob_service_client.get_blob_client(container=BLOB_CONTAINER, blob=BLOB_NAME)
    next_time = datetime.now() + timedelta(days=30)
    blob_client.upload_blob(next_time.isoformat(), overwrite=True)

def main(timer):
    current_time = datetime.now()
    next_monthly_time = get_next_monthly_time()

    if current_time >= next_monthly_time:
        # 执行月度任务
        print("执行月度任务")
        # 更新下次执行时间
        update_next_monthly_time()
    else:
        # 执行每日任务
        print("执行每日任务")

注意：需要提前在Azure Storage中创建名为job-markers的Blob容器，或者在代码中添加容器不存在时自动创建的逻辑。

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