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)