如何在 Django 视图中使用 APscheduler 安排用户定义的作业
How to schedule user-defined jobs with APscheduler in a Django view
这个想法是允许用户向服务器提交 time/date 以安排作业在用户指定的时间执行。
我的第一选择是在视图中执行此操作,前端的数据将发布到该视图。但是,apscheduler returns 一个值错误。
下面是我希望用于用户安排的作业和一个小测试的视图的片段。
class SchedulePolicyDeployView(APIView):
def post(self, request, version):
def test_job():
print("cron job test")
scheduler.start()
scheduler.add_job(test_job, "cron", id="test_job", day="*", minute="*/1")
return Response("job scheduled successfully")
返回的错误信息是:
ValueError: This Job cannot be serialized since the reference to its callable (<function SchedulePolicyDeployView.post.<locals>.test_job at 0x05694C00>) could not be determined. Consider giving a textual reference (module:function name) instead.
所以我确实在我的代码中修复了这个问题;但自从我问了这个问题后,情况发生了很大变化。不过,我已尝试将所有内容都放在此处:
*假设调度器对象已经初始化
scheduler.py
def schedule_job(arg1, arg2, arg3, time, job_id):
scheduler.add_job(deploy_policy, 'date', args=[arg1, arg2, arg3], run_date=datetime(int(date['year']), int(date['month']), int(date['day']), int(time['hour']), int(time['minute']), 0))
return Response({"message": "scheduled!"})
views.py
from scheduler import schedule_job
class SchedulePolicyDeployView(APIView):
def post(self, request, version):
schedule_job(apic, username, password, policy_url_and_payload_result, date_dict, time_dict, payload["job_id"])
return Response({"result": "success"})
这个想法是允许用户向服务器提交 time/date 以安排作业在用户指定的时间执行。
我的第一选择是在视图中执行此操作,前端的数据将发布到该视图。但是,apscheduler returns 一个值错误。
下面是我希望用于用户安排的作业和一个小测试的视图的片段。
class SchedulePolicyDeployView(APIView):
def post(self, request, version):
def test_job():
print("cron job test")
scheduler.start()
scheduler.add_job(test_job, "cron", id="test_job", day="*", minute="*/1")
return Response("job scheduled successfully")
返回的错误信息是:
ValueError: This Job cannot be serialized since the reference to its callable (<function SchedulePolicyDeployView.post.<locals>.test_job at 0x05694C00>) could not be determined. Consider giving a textual reference (module:function name) instead.
所以我确实在我的代码中修复了这个问题;但自从我问了这个问题后,情况发生了很大变化。不过,我已尝试将所有内容都放在此处:
*假设调度器对象已经初始化
scheduler.py
def schedule_job(arg1, arg2, arg3, time, job_id):
scheduler.add_job(deploy_policy, 'date', args=[arg1, arg2, arg3], run_date=datetime(int(date['year']), int(date['month']), int(date['day']), int(time['hour']), int(time['minute']), 0))
return Response({"message": "scheduled!"})
views.py
from scheduler import schedule_job
class SchedulePolicyDeployView(APIView):
def post(self, request, version):
schedule_job(apic, username, password, policy_url_and_payload_result, date_dict, time_dict, payload["job_id"])
return Response({"result": "success"})