当前位置：首页 > news >正文

多数据库使用django-apscheduler时，migrate后并不能生成django_apscheduler_djangojob表的问题

news 2025/8/9 21:05:46

先说一下django-apscheduler定时器的使用过程：

django-apscheduler基本使用

1.安装django-apscheduler代码如下（示例）：

pip install django-apscheduler

2.配置settings.py的INSTALLED_APPS代码如下（示例）：

INSTALLED_APPS = (# ..."django_apscheduler",
)

3.通过命令生成定时记录表

使用python的manage命令，我们应该cd到项目目录下，就是manage.py文件所在的目录。

我们应该使用python manage.py makemigrations 和 python manage.py migrate同步数据库,数据库中将生成2个表，django_apscheduler_djangojob和django_apscheduler_djangojobexecution。

这里有问题！！!migrate后，死活没有在MySQL中创建django_apscheduler_djangojob和django_apscheduler_djangojobexecution表

问题原因：

因为我的项目是多数据库配置的，在配置settings.py的INSTALLED_APPS代码后，要在DATABASE_APPS_MAPPING中也要配置一下django_apscheduler应该映射到哪个数据库生成表。

解决办法：

DATABASE_APPS_MAPPING = {........这里的代码应该有default的数据库映射的代码了'django_apscheduler': 'default',  ##主要是就是这行代码
}

完事再来一次：

python manage.py makemigrations
python manage.py migrate

成功生成以下数据表！

4.创建runapscheduler.py文件

runapscheduler.py正是通过自定义创建manange命令的py文件，可以通过python manage.py runapscheduler 启动定时程序。

import loggingfrom django.conf import settingsfrom apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from django.core.management.base import BaseCommand
from django_apscheduler.jobstores import DjangoJobStore
from django_apscheduler.models import DjangoJobExecution
from django_apscheduler import utillogger = logging.getLogger(__name__)def my_job():# Your job processing logic here...print(123)passdef delete_old_job_executions(max_age=604_800):"""此作业从数据库中删除早于“max_age”的APScheduler作业执行条目。它有助于防止数据库中塞满不再有用的旧历史记录。最长7天"""DjangoJobExecution.objects.delete_old_job_executions(max_age)class Command(BaseCommand):help = "Runs APScheduler."def handle(self, *args, **options):scheduler = BlockingScheduler(timezone=settings.TIME_ZONE)scheduler.add_jobstore(DjangoJobStore(), "default")scheduler.add_job(my_job,trigger=CronTrigger(second="*/10"),  # Every 10 secondsid="my_job",  # The `id` assigned to each job MUST be uniquemax_instances=1,replace_existing=True,)logger.info("Added job 'my_job'.")scheduler.add_job(delete_old_job_executions,trigger=CronTrigger(day_of_week="mon", hour="00", minute="00"),  # Midnight on Monday, before start of the next work week.id="delete_old_job_executions",max_instances=1,replace_existing=True,)logger.info("Added weekly job: 'delete_old_job_executions'.")try:logger.info("Starting scheduler...")scheduler.start()except KeyboardInterrupt:logger.info("Stopping scheduler...")scheduler.shutdown()logger.info("Scheduler shut down successfully!")

1234这4步是django-apscheduler官网的使用步骤，经过测试，定时程序已经正常运行了。