23π
Well, Iβll have a go. Assuming you have installed apscheduler (or put it in your Python path) The basic documentation for APS lists the following code for starting up a job:
from apscheduler.scheduler import Scheduler
sched = Scheduler()
@sched.interval_schedule(hours=3)
def some_job():
print "Decorated job"
sched.configure(options_from_ini_file)
sched.start()
The issue I suspect your having is how to trigger the startup of that in Django. You have multiple options, mostly consisting of βputting it somewhere it will be runβ. For example, add it to the bottom of a models.py
file, or in urls.py
and it will be processed once on each Django restart, then continue to run in the background. Database accesses can then be performed as normal from within the function β just import your models as normal for your queries.
However, consider that if you do this youβre going to need to kill & restart your server any time you want to alter the scheduling. Iβm also unsure how this is going to fare with multiple threads (will you have >1 timer?)
You really might want to consider going with something like Celery (via django-celery) which gets you all this, plus a separate scheduling daemon with finer control of scheduling.
7π
To use this apscheduler, you can set it up with the code below, the only part you may be having difficulties with is using the jobstore sqlalchemy
is recommended but since you already have a django ORM, that can work also.
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
scheduler.start()
def myjob():
print('hello')
scheduler.add_job(myjob, 'cron', hour=0)
scheduler.add_job(
myjob,
'date',
id=id_value,
jobstore="default",
run_date=scheduler_date,
replace_existing=True
)
I would recommend using : django-apscheduler which already integrate with jobstore with DjangoORM and also provides a web interface to manage jobs
For django-apscheduler, you can use the code below
import time
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_events,
register_job
scheduler = BackgroundScheduler()
scheduler.add_jobstore(DjangoJobStore(), "default")
@register_job(scheduler, "interval", seconds=1)
def test_job():
time.sleep(4)
print("I'm a test job!")
# raise ValueError("Olala!")
register_events(scheduler)
scheduler.start()
print("Scheduler started!")
- Accessing django project in LAN systems
- Changes made to (static) CSS file not reflecting in Django development server
4π
just try below code.
project name
: crawling_server
django-admin startproject crawling_server
cd crawling_server
app name
: stock_crawling
python manage.py startapp stock_crawling
purpose
: print βhello worldβ every second
1. install packages
pip install -r requirements.txt
APScheduler==3.8.1
asgiref==3.4.1
backports.zoneinfo==0.2.1
beautifulsoup4==4.10.0
certifi==2021.10.8
Django==3.2.10
django-apscheduler==0.6.0
pytz==2021.3
pytz-deprecation-shim==0.1.0.post0
six==1.16.0
soupsieve==2.3.1
sqlparse==0.4.2
typing_extensions==4.0.1
tzdata==2021.5
tzlocal==4.1
2.
add below code at settings.py
INSTALLED_APPS = [
...
'stock_crawling',
'django_apscheduler',
]
APSCHEDULER_DATETIME_FORMAT = "N j, Y, f:s a" # Default
SCHEDULER_DEFAULT = True
3.
add below code at views.py
import datetime
def send_hello():
time = datetime.datetime.now()
print('hello world:[{}]'.format(time))
4
generate operator.py
in project dir (see below dir tree)
./
βββ crawling_server <- **project dir**
β βββ asgi.py
β βββ operator.py <- **generate this script**
β βββ settings.py
β βββ urls.py
β βββ wsgi.py
βββ db.sqlite3
βββ manage.py
βββ requirements.txt
βββ stock_crawling
βββ admin.py
βββ apps.py
βββ migrations
βββ models.py
βββ tests.py
βββ views.py
add below code at operator.py
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import register_events, DjangoJobStore
from stock_crawling.views import send_hello
def start():
scheduler = BackgroundScheduler()
scheduler.add_jobstore(DjangoJobStore(), 'djangojobstore')
register_events(scheduler)
@scheduler.scheduled_job('interval', seconds=1, name='auto_hello')
def auto_hello():
send_hello()
scheduler.start()
Either interval or cron is used as the scheduler trigger. For more information, see the link below.
example interval trigger
@scheduler.scheduled_job('interval', seconds=1, name='auto_hello')
example cron trigger
@scheduler.scheduled_job('cron', hour=1, name='auto_hello')
5.
add below code at app.py
in project dir
from django.apps import AppConfig
from django.conf import settings
class StockCrawlingConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'stock_crawling'
def ready(self):
if settings.SCHEDULER_DEFAULT:
from crawling_server import operator
operator.start()
6. runserver
python manage.py runserver --noreload
7. result
hello world:[2021-12-13 12:34:33.982562]
Job '259a33f9664248bea607aa9ba0cdd587' no longer exists! Skipping logging of job execution...
Job '259a33f9664248bea607aa9ba0cdd587' no longer exists! Skipping logging of job execution...
hello world:[2021-12-13 12:34:34.982743]
Job '259a33f9664248bea607aa9ba0cdd587' no longer exists! Skipping logging of job execution...
Job '259a33f9664248bea607aa9ba0cdd587' no longer exists! Skipping logging of job execution...
hello world:[2021-12-13 12:34:35.986722]
- Django REST Framework: define fields in nested object?
- Where to instantiate boto s3 client so it is reused during a request?
- Django β show loading message during long processing
- Django: How to check if something is an email without a form