Python+Heroku环境下Flask项目定时推送通知失效问题求助

看起来你在Flask项目里用APScheduler实现定时推送的环节踩了坑——手动触发推送正常，但定时任务就是不跑，确实挺闹心的。我帮你梳理几个最可能的问题点，以及对应的解决办法：

1. 调度器初始化时机与调试模式冲突 #

你当前的app.py里，APScheduler的初始化和启动是在加载环境变量、配置Flask之前完成的，这会导致调度器无法正确读取Flask的配置。更关键的是，开启DEBUG模式时Flask默认会启动两个进程（主进程+重载进程），这会造成调度器被重复初始化，甚至原进程的任务被覆盖或丢失。

修复方案： #

• 调整初始化顺序，先加载配置再启动调度器；
• 调试模式下关闭自动重载：

# app.py 调整后的初始化流程
from flask import Flask
# ... 其他导入

app = Flask(__name__)
load_dotenv(".env")  # 先加载环境变量

# 配置Flask参数
app.config['DEBUG'] = True
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL', 'sqlite:///data.db')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# ... 其他配置项

# 最后初始化并启动调度器
scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()

# ... 中间的路由、资源注册代码

if __name__ == '__main__':
    from db import db
    db.init_app(app)
    db.app = app
    ma.init_app(app)
    if app.config['DEBUG']:
        @app.before_first_request
        def create_tables():
            db.create_all()
    # 关闭调试模式的自动重载
    app.run(port=5000, use_reloader=False)

2. 定时任务的时间格式/时区问题 #

你把时间戳转成字符串'%Y-%m-%d %H:%M:%S'传给run_date，APScheduler对字符串格式的解析可能存在时区或格式匹配问题。直接传入datetime对象能避免这类解析错误。

修复方案： #

修改schedule_notif函数，直接传递datetime对象：

def schedule_notif(timestamp, token, n_id):
    # 直接生成datetime对象，无需转字符串
    run_date = datetime.fromtimestamp(timestamp)
    # 如果你的时间戳是UTC时区，建议显式指定：
    # import pytz
    # run_date = datetime.utcfromtimestamp(timestamp).replace(tzinfo=pytz.UTC)
    
    app.apscheduler.add_job(
        id=n_id, 
        func=send_push, 
        trigger='date', 
        run_date=run_date, 
        args=(token, "You have 30 minutes until your booking", "You have 30 minutes until your booking", n_id)
    )

3. 内存存储导致任务丢失 #

APScheduler默认使用内存存储任务，一旦Flask进程重启（比如调试重载、服务器重启），所有未执行的定时任务都会丢失。即使是开发环境，调试模式的自动重载也会清空任务列表。

临时解决方案（开发阶段）： #

保持use_reloader=False，避免进程重载。

生产环境解决方案： #

配置APScheduler使用持久化存储，比如Redis或数据库：

# 在Flask配置中添加
app.config['SCHEDULER_JOBSTORES'] = {
    'default': {
        'type': 'redis',
        'url': os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
    }
}

4. 任务函数的上下文与日志缺失 #

你的send_push函数如果触发异常，APScheduler不会主动把错误返回给客户端，也不会默认记录日志，导致你无法知道任务是否执行、执行时有没有报错。另外，如果任务中需要访问Flask资源（比如数据库），还需要手动激活应用上下文。

修复方案： #

给send_push添加日志记录，必要时激活上下文：

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def send_push(token, title, message, n_id):
    push_service = FCMNotification(api_key=api_key)
    try:
        # 如果需要访问Flask资源（比如数据库），需要激活上下文
        # with app.app_context():
        #     相关操作
        registration_id = token.token
        if registration_id:
            message_title = title
            message_body = message
            data = {
                "title": title,
                "body": message,
                "type": "booking_reminder"
            }
            result = push_service.notify_single_device(
                registration_id=registration_id, 
                message_title=message_title, 
                message_body=message_body, 
                data_message=data, 
                timeout=30
            )
            logger.info(f"Notification {n_id} sent successfully: {result}")
    except Exception as err:
        logger.error(f"Failed to send notification {n_id}: {str(err)}")

5. 开启调度器日志排查 #

开启APScheduler的日志，能直观看到任务是否被添加、触发时间是否正确、执行时有没有异常：

# 在app.py开头添加
import logging
logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.INFO)

按以上步骤逐一排查，应该能定位到定时任务不执行的原因。

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