Celery beat 사용

주기적으로 코드를 확인하는 기능을 구현하기 위해서 celery beat를 사용했다.

(celery를 사용하기 위한 기본 설정은 이전 글에 있으니 참고하시면 됩니다.)

 

#config/celery.py
...
..
app.conf.timezone = 'Asia/Seoul'

app.conf.beat_schedule = {
    'test_periodic_task': {
        'task': '앱이름.tasks.task이름',
        'schedule': crontab(minute='*/1'), #timedelta(seconds=10)  
    },
}
#minute=1 그냥 이렇게 적으면 매 시간의 1분에 동작하는 것. 8시1분 9시1분 이런식

 

예시 tasks.py입니다.

from celery import shared_task

@shared_task
def test_periodic_task():
    print("test Periodic task")
    return "Complete"

 

---

실행하는 과정

celery -A 프로젝트이름 beat -l INFO

위의 코드를 사용하면 beat를 실행할 수 있다.

---

더 자세하게 확인을 하고 싶다면 아래 코드 사용

celery -A 프로젝트이름 beat -l debug

---

celery를 admin에 추가를 하고 싶다면 아래 코드를 사용

celery -A 프로젝트이름 beat -l INFO --scheduler django_celery_beat.schedulers:DatabaseScheduler

 

 

---

https://backstreet-programmer.tistory.com/198

[Django + Redis + Celery] Python 분산 비동기 작업 큐 (Distributed Task/Job Queue) 튜토리얼

 

https://pypi.org/project/django-celery-beat/

django-celery-beat

 

공부용으로 작성한 글이며 문제시 삭제하고 틀린부분이 있다면 피드백 부탁드려요!