Django - run a function every x seconds
I recommend you use Celery's task management. You can refer this to set up this app (package if you're from javaScript background).
Once set, you can alter the code to:
@app.task
def check_shut_down():
if not some_fun():
# add task that'll run again after 2 secs
check_shut_down.delay((), countdown=3)
else:
# task completed; do something to notify yourself
return True
This answer expands on Oz123's answer a little bit.
In order to get things working, I created a file called mainapp/jobs.py
to contain my scheduled jobs. Then, in my apps.py
module, I put from . import jobs
in the ready
method. Here's my entire apps.py
file:
from django.apps import AppConfig
import os
class MainappConfig(AppConfig):
name = 'mainapp'
def ready(self):
from . import jobs
if os.environ.get('RUN_MAIN', None) != 'true':
jobs.start_scheduler()
(The RUN_MAIN
check is because python manage.py runserver
runs the ready
method twice—once in each of two processes—but we only want to run it once.)
Now, here's what I put in my jobs.py
file. First, the imports. You'll need to import Scheduler
, threading
and time
as below. The F
and UserHolding
imports are just for what my job does; you won't import these.
from django.db.models import F
from schedule import Scheduler
import threading
import time
from .models import UserHolding
Next, write the function you want to schedule. The following is purely an example; your function won't look anything like this.
def give_admin_gold():
admin_gold_holding = (UserHolding.objects
.filter(inventory__user__username='admin', commodity__name='gold'))
admin_gold_holding.update(amount=F('amount') + 1)
Next, monkey-patch the schedule
module by adding a run_continuously
method to its Scheduler
class. Do this by using the below code, which is copied verbatim from Oz123's answer.
def run_continuously(self, interval=1):
"""Continuously run, while executing pending jobs at each elapsed
time interval.
@return cease_continuous_run: threading.Event which can be set to
cease continuous run.
Please note that it is *intended behavior that run_continuously()
does not run missed jobs*. For example, if you've registered a job
that should run every minute and you set a continuous run interval
of one hour then your job won't be run 60 times at each interval but
only once.
"""
cease_continuous_run = threading.Event()
class ScheduleThread(threading.Thread):
@classmethod
def run(cls):
while not cease_continuous_run.is_set():
self.run_pending()
time.sleep(interval)
continuous_thread = ScheduleThread()
continuous_thread.setDaemon(True)
continuous_thread.start()
return cease_continuous_run
Scheduler.run_continuously = run_continuously
Finally, define a function to create a Scheduler
object, wire up your job, and call the scheduler's run_continuously
method.
def start_scheduler():
scheduler = Scheduler()
scheduler.every().second.do(give_admin_gold)
scheduler.run_continuously()
For many small projects celery is overkill. For those projects you can use schedule, it's very easy to use.
With this library you can make any function execute a task periodically:
import schedule
import time
def job():
print("I'm working...")
schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
while True:
schedule.run_pending()
time.sleep(1)
The example runs in a blocking manner, but if you look in the FAQ, you will find that you can also run tasks in a parallel thread, such that you are not blocking, and remove the task once not needed anymore:
import threading
import time
from schedule import Scheduler
def run_continuously(self, interval=1):
"""Continuously run, while executing pending jobs at each elapsed
time interval.
@return cease_continuous_run: threading.Event which can be set to
cease continuous run.
Please note that it is *intended behavior that run_continuously()
does not run missed jobs*. For example, if you've registered a job
that should run every minute and you set a continuous run interval
of one hour then your job won't be run 60 times at each interval but
only once.
"""
cease_continuous_run = threading.Event()
class ScheduleThread(threading.Thread):
@classmethod
def run(cls):
while not cease_continuous_run.is_set():
self.run_pending()
time.sleep(interval)
continuous_thread = ScheduleThread()
continuous_thread.setDaemon(True)
continuous_thread.start()
return cease_continuous_run
Scheduler.run_continuously = run_continuously
Here is an example for usage in a class method:
def foo(self):
...
if some_condition():
return schedule.CancelJob # a job can dequeue it
# can be put in __enter__ or __init__
self._job_stop = self.scheduler.run_continuously()
logger.debug("doing foo"...)
self.foo() # call foo
self.scheduler.every(5).seconds.do(
self.foo) # schedule foo for running every 5 seconds
...
# later on foo is not needed any more:
self._job_stop.set()
...
def __exit__(self, exec_type, exc_value, traceback):
# if the jobs are not stop, you can stop them
self._job_stop.set()
I can't comment on oz123's (https://stackoverflow.com/a/44897678/1108505) and Tanner Swett's (https://stackoverflow.com/a/60244694/5378866) excellent post, but as a final note I wanted to add that if you use Gunicorn and you have X number of workers, the section:
from django.apps import AppConfig
import os
class MainappConfig(AppConfig):
name = 'mainapp'
def ready(self):
from . import jobs
if os.environ.get('RUN_MAIN', None) != 'true':
jobs.start_scheduler()
will be executed that same number of times, launching X schedulers at the same time.
If we only want it to run only one instance (for example if you're going to create objects in the database), we would have to add in our gunicorn.conf.py file something like this:
def on_starting(server):
from app_project import jobs
jobs.start_scheduler()
And finally in the gunicorn call add the argument --preload