如何定期运行带参数的自定义函数，尝试apscheduler无法执行怎么办

1 先修复现有代码的显性错误 #

你的myFunction类中存在未定义变量，会直接导致调用失败：

class myFunction():
    def __init__(self, n_user, n_password, n_url, n_port, db_password):
        self.zaman = datetime.now().strftime("%Y-%m-%d")
        # 原代码的a_url未定义，改为传入的n_url参数
        self.location = n_url
        self.static_fields = {}
        self.index_name = "vulns-" + self.zaman
        self.download_all(n_user, n_password, n_url, n_port, db_password)
        self.send_it()

2 基于django-apscheduler实现定时任务 #

推荐使用适配Django的django-apscheduler库，它会自动持久化任务记录，避免重复执行问题。

2.1 安装依赖 #

pip install django-apscheduler

2.2 配置Django项目 #

在settings.py的INSTALLED_APPS中新增注册：

INSTALLED_APPS = [
    # ... 其他原有app
    'django_apscheduler',
]

执行迁移生成任务记录表：

python manage.py migrate

3 编写定时任务逻辑 #

在你的app目录下新建scheduler.py，定义任务逻辑：

from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_events
from .models import DashboardData
# 导入你的myFunction类
from .functions import myFunction
import datetime

def run_my_function_for_all_users():
    """遍历所有DashboardData记录，逐个传入参数调用myFunction"""
    all_configs = DashboardData.objects.all()
    for config in all_configs:
        # 取出对应参数传入类初始化
        try:
            myFunction(
                n_user=config.n_username,
                n_password=config.n_password,
                n_url=config.n_url,
                n_port=config.n_port,
                db_password=config.db_password
            )
        except Exception as e:
            # 可自行添加日志记录错误，避免单个配置出错导致整个任务崩溃
            print(f"用户{config.user.id}的任务执行失败: {str(e)}")

def start_scheduler():
    scheduler = BackgroundScheduler(timezone='Asia/Shanghai') # 替换为你所在的时区
    scheduler.add_jobstore(DjangoJobStore(), "default")
    # 添加15分钟间隔的定时任务
    scheduler.add_job(
        run_my_function_for_all_users,
        'interval',
        minutes=15,
        id='my_function_periodic_task', # 唯一任务ID
        replace_existing=True, # 重复启动时覆盖已有同ID任务
    )
    register_events(scheduler)
    scheduler.start()

4 启动调度器 #

在你的app的apps.py中，项目启动时自动加载调度器：

from django.apps import AppConfig
import os

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app_name' # 替换为你的app名称

    def ready(self):
        # 避免Django调试模式下双进程重复启动调度器
        if os.environ.get('RUN_MAIN') == 'true':
            from .scheduler import start_scheduler
            start_scheduler()

常见问题排查 #

• 调试阶段启动服务时使用python manage.py runserver --noreload关闭重载，避免调度器重复启动
• 如果需要使用DashboardData中的period字段自定义每个用户的执行间隔，可以改为在新增/修改DashboardData记录时动态添加/修改对应任务即可
• 生产环境部署时，建议单独启动一个进程运行调度器，避免和web服务进程耦合

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