If you need to run a function every X minutes/seconds etc. to do some cleanups, trigger some operations you can do a simple scheduler with the help of threading module and django custom cli commands.
Let's say I want to invoke a function every 5 seconds to post something on an external API.
In your django app create a folder/package named management
inside that folder create another folder named commands
. In the commands
folder create a module named runposter.py
. In the end you'll have something like this structure yourapp/management/commands/runposter.py
.
In this code we use a thread which runs a while loop as long as it's not stopped every 5 seconds. Replace print("posting")
with the function/logic you want to run.
# runposter.py
import time
from threading import Thread, Event
from django.conf import settings
from django.core.management.base import BaseCommand
stop_event = Event()
def my_job():
while not stop_event.is_set():
try:
print("posting")
time.sleep(5)
except KeyboardInterrupt:
break
except Exception as err:
print(err)
continue
class Command(BaseCommand):
help = "Run Poster."
def handle(self, *args, **options):
poster = Thread(target=my_job)
try:
print("Starting poster...")
poster.start()
while poster.is_alive():
poster.join(timeout=1)
except KeyboardInterrupt:
print("Stopping poster...")
stop_event.set()
poster.join()
print("Poster shut down successfully!")
Nice, now open another terminal window and run python manage.py runposter
. Command runposter as you can see was created from the module name we've given.
Of course, for something more complex I recommend using rq-scheduler or celery periodic tasks or django-q.
But, for simple cases this should be good enough.
Top comments (0)