Django集成APScheduler报错：Expected a job store instance or a string, got NoneType

问题根源 #

1. API调用错误：你把添加定时任务的scheduler.add_job()写成了配置任务存储的scheduler.add_jobstore()，而且直接调用了self.send_email_to_registered_users()——这个函数执行后返回None，传入add_jobstore就触发了TypeError。
2. 无异常处理：邮件发送时如果遇到无效邮箱、网络故障等问题，会直接抛出异常中断任务，导致整个调度器停止运行。

修复后的代码 #

import logging
from django.conf import settings
from apscheduler.schedulers.blocking import BlockingScheduler
from django.core.management.base import BaseCommand
from django.core.mail import EmailMessage
from ...models import Assessment
from django_apscheduler.jobstores import DjangoJobStore
from apscheduler.triggers.cron import CronTrigger


logger = logging.getLogger(__name__)


class Command(BaseCommand):
    help = "Running in the dark :)"

    def send_email_to_registered_users(self):
        mail_subject = "Newsletter"
        message = "Welcome to our newsletter"
        
        # 分批获取数据，避免内存过载
        for entry in Assessment.objects.iterator():
            sender_email = entry.email
            if not sender_email:
                logger.warning(f"跳过条目 {entry.id}：无邮箱地址")
                continue
                
            try:
                email_send = EmailMessage(mail_subject, message, to=[sender_email])
                email_send.send()
                logger.info(f"邮件已发送至 {sender_email}")
            except Exception as e:
                logger.error(f"发送邮件至 {sender_email} 失败：{str(e)}")
        
        logger.info("本轮邮件发送完成")

    def handle(self, *args, **kwargs):
        scheduler = BlockingScheduler(timezone=settings.TIME_ZONE)
        # 配置Django任务存储
        scheduler.add_jobstore(DjangoJobStore(), "default")
        
        # 添加定时任务：传函数对象而非调用函数（不要加括号）
        scheduler.add_job(
            self.send_email_to_registered_users,
            trigger=CronTrigger(second="*/10"),
            id="send_email_to_registered_users",
            max_instances=10,
            jobstore="default"
        )
        
        logger.info("调度器已启动，任务开始运行...")
        try:
            scheduler.start()
        except (KeyboardInterrupt, SystemExit):
            logger.info("手动停止调度器")
            scheduler.shutdown()

关键修改说明 #

• 修正API调用：用add_job()替代add_jobstore()添加定时任务，传入函数对象而非直接调用函数。
• 异常捕获：给单条邮件发送逻辑加try-except，单个邮件失败不会导致整个任务崩溃。
• 优化数据查询：用iterator()分批获取数据库记录，避免一次性加载大量数据占用内存。
• 日志完善：添加详细日志，方便排查发送失败的具体情况。
• 优雅关闭：捕获手动停止信号，确保调度器正常关闭。

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