django-crontab을 사용하려고 했는데 리눅스 기반이라 윈도우에서는 못쓴다고 한다.
APScheduler 사용하기
https://hello-bryan.tistory.com/216
$ pip install APScheduler
1. Blocking Scheduler
Blocking Scheduler를 쓸 것임
views.py에 작성한 테스트 코드
urls.py에서 scheduler_test 함수를 연결해줘야 실행된다. (연결안하고 파이썬 코드 그대로 갖다놨더니 뭔가 엄청 오래 걸렸음)
연결을 하고나니 scheduler_test() takes 0 positional arguments but 1 was given 이라는 오류가 떴는데 대체 어디서 파라미터가 들어갔는지 모르겠어서 인자로 self를 넣어주니까 해결됐다.
실행결과
1분에 한번씩 job 함수가 실행된다. 그리고 Blocking Scheduler이기 때문에 sched.start() 뒤에 있는 sched after ~ 는 출력되지 않는 것을 볼 수 있다. 그리고 url에 html 파일을 연결해놓지 않고 함수만 연결해놔서 터미널에서 파이썬 코드가 돌아가는 것은 확인할 수 있었지만 웹페이지는 runserver를 끊을 때까지 계속 로딩 화면이었다.
2. Background Scheduler
아니 근데 Blocking Scheduler를 쓰려고 했는데 주어진 태스크를 보면 'Django 에서 runserver 했을 때, 주기적으로 특정 API 를 실행할 수 있는 방법' 이다. 지금 방법으로는 1분마다 불러오다가 다른 페이지로 넘어가면 불러오는게 중지된다. 이거를 꺼도 계속 주기적으로 받아오려면 Background Scheduler를 써야하나보다.
Blocking 코드에서 Blocking을 Background로 바꾼다.
기묘하게 실행된.... 실행결과
sched before & after 까지 다 출력된 다음에 에러메시지 실컷 띄워놓고 스케쥴링이 계속되었다.
scheduler_test 함수 마지막에 이 부분을 추가하면?
from django.shortcuts import render
import time
from apscheduler.schedulers.background import BackgroundScheduler
def job():
print(f'scheduler testing : {time.strftime("%H:%M:%S")}')
def scheduler_test(self):
sched = BackgroundScheduler()
sched.add_job(job, 'cron', second='0', id='test')
print('sched before ~')
sched.start()
print('sched after ~')
# 이 부분!
while True:
time.sleep(1)
def next_job(self):
print('next job')
페이지를 중간에 바꾸고(next job 출력) 다시 scheduler test로 돌아오지 않았는데 스케쥴링이 계속된다. 위에서 나왔던 Internal Server Error : / 도 안떴다. (이번에 뜬건 next job 에러)
그럼 지금까지 나온거를 정리하면 Background Scheduler를 이용해서 다른 함수를 실행할 때도 백그라운드에서 주기적으로 호출해서 태스크를 수행할 수 있다. 현재
job자리에 날씨 API를 호출하는 함수를 넣으면 되겠다. 그럼 남은거는runserver를 돌릴 때 자동으로 이 함수가 호출되도록 하는 거
서버 시작할 때 특정 코드 실행하기
이때 쓰는게 AppConfig 라고 한다.
# app/views.py
from django.apps import AppConfig
class MyAppConfig(AppConfig):
name = 'MY_APP_NAME'
def ready(self):
# TODO: 서버 시작할 때 실행할 코드# app/__init__.py
default_app_config = 'app.broker.MyAppConfig'이걸 쓸때는 두번 실행되는 일을 피하기 위해 장고 서버를 켤 때 --noreload 옵션을 넣는다.
python manage.py runserver --noreload테스트를 위해 앞에서 scheduler_test 에 연결해뒀던 url을 제거하고 실행해보았다.
# app/views.py
from django.shortcuts import render
import time
from apscheduler.schedulers.background import BackgroundScheduler
from django.apps import AppConfig
def job():
print(f'scheduler testing : {time.strftime("%H:%M:%S")}')
def scheduler_test(self):
sched = BackgroundScheduler()
sched.add_job(job, 'cron', second='0', id='test')
print('sched before ~')
sched.start()
print('sched after ~')
while True:
time.sleep(1)
def next_job(self):
print('next job')
class MyAppConfig(AppConfig):
name = "api"
def ready(self):
scheduler_test(self)# app/__init__.py
default_app_config = 'api.views.MyAppConfig'runserver 돌리자마자 scheduler 테스트 코드가 실행되었다.
ERROR
근데 문제가 생김...
runserver를 돌렸는데 스케쥴러 시작은 되는데 127.0.0.1:8000 에 연결이 안된다.
Appconfig에 등록을 했으면 그게 한번만 실행되고 서버가 실행이 되는 순선데
scheduler_test를 한번만 실행되게 해놨다. 이게 무슨 뜻이냐면 주기적으로 불러오는 스케쥴러를 한번 실행한다. 즉 이 한번 실행하는 이 태스크가 영원히 안끝나는거라는 뜻......ready()에 단순 프린트문을 찍었더니 이렇게 제대로 나왔다.
그럼 어떻게 해결하지? 일단 한번 부르고 나중에 주기적으로 불러야 할듯싶다. 근데 Background 스케줄러인데 왜 이러는지 모르겠다..
장고 프로젝트에 이식하기
일단 여기까지 하고 프로젝트에 이식해보도록 하겠다!
파일구조는 이렇게 생겼고 weather_api.py 에 날씨 데이터를 가져오는 함수(check_weather())가 있다. 나는 views.py 에서 이 함수를 import 시켜서 1분에 한번씩 API 호출을 해보도록 하겠다. 일단 Django라는건 무시하고 runserver가 아닌 파이썬 파일을 실행시키는 것으로 간주
코드
# api/views.py
from django.shortcuts import render
from apscheduler.schedulers.background import BackgroundScheduler
import time
from weather_api import check_weather
# Create your views here.
def job():
print(f'******{time.strftime("%H:%M:%S")}******')
check_weather()
print("************************")
def cron_weather():
sched = BackgroundScheduler()
sched.add_job(job, 'cron', second='0', id='cron_weather')
sched.start()
while True:
time.sleep(1)
cron_weather()# api/weather_api.py
def check_weather():
url = "https://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtFcst"
# ~ 생략 ~
print("response: ", weather_data)
return weather_data실행결과
1분에 한번씩 check_weather() 함수를 호출해서 날씨 정보를 받아오고 있다. (API 특성상 30분마다 정보가 업데이트가 돼서 계속 똑같은 정보가 찍히긴 한다.)
