[Django][Django Apscheduler] Django 에서 Scheduler 사용하기

내가 실무를 해본적은 없지만, 문득 db에 데이터만 쌓이면 너무 db가 무거워질거란 생각이 들어 주기적으로 정리를 해야겠단 생각으로 

Apscheduler 를 추가해보았다.

나는 데이터를 삭제를 했는데 근데 지금 생각으로 삭제보다는 다른 db에 저장하는게 낫지 않았을까 싶기도 하다! 데이터를 쉽게 날리는건 많은 고민이 필요하기 때문!

 

아무튼 그래서 그 때는 그랬다! 정리할 것은 2개

1. 공사 작업이 완료된 시점으로 30일이 지난 작업

2. VOC가 확인완료 상태인 시점으로 1일 후인 객체

 

그리고 나는 30초마다 확인하도록 했는데 이건 너무 소모가 큰것같다. 30초보다 요일을 정해서 실행시키는 것이 더 나을 듯한다. 

 

 

---

먼저, 해당 라이브러리를 등록시켜야한다.

* APSCHEDULER_RUN_NOW_TIMEOUT:  타임아웃(최대 실행 시간) 설정값이고, 디폴트가 25초이다.

* SCHEDULER_DEFAULT: 애플리케이션의 스케줄러를 기본적으로 시작할지 여부를 결정. True로 설정되어 있으므로 애플리케이션이 시작될 때 스케줄러가 자동으로 시작됨

 

[settings.py]

# Application definition
INSTALLED_APPS = [
	. . .
    "django_apscheduler",
]

CSRF_COOKIE_HTTPONLY = True

APSCHEDULER_DATETIME_FORMAT = "N j, Y, f:s a"
APSCHEDULER_RUN_NOW_TIMEOUT = 25  
SCHEDULER_DEFAULT = True

 

---

그리고 Job을 등록한다.

1. delete_completed_constructions() : Post 모델에서 30일이 지난 완료된 작업을 삭제하는 함수
2. delete_confirmed_voc() : VOC 모델에서 1일 전에 업데이트된 '확인완료' 상태의 VOC 객체를 삭제하는 함수
3. start() : 백그라운드 스케줄러를 생성하고 위의 두 함수를 주기적으로 실행

[Gongsa/runapscheduler.py]

import logging
from django.utils import timezone
from datetime import datetime, timedelta
from .models import Post
from aici.models import VOC
from apscheduler.schedulers.background import BackgroundScheduler
from django.conf import settings
from apscheduler.triggers.cron import CronTrigger

logger = logging.getLogger(__name__)

def delete_completed_constructions():
    thirty_days_after_end = datetime.now().date() - timedelta(days=30)
    constructions_to_delete = Post.objects.filter(end_at__lte=thirty_days_after_end)
    constructions_to_delete.delete()

def delete_confirmed_voc():
    one_day_ago = timezone.now() - timedelta(days=1)
    confirmed_voc = VOC.objects.filter(voc_status='확인완료', updated_at__lte=one_day_ago)
    confirmed_voc.delete()

def start():
    scheduler = BackgroundScheduler(timezone=settings.TIME_ZONE)

    scheduler.add_job(
        delete_completed_constructions,
        trigger=CronTrigger(second="*/30"),  #30초마다 실행
        id="my_job",
        max_instances=1,
        replace_existing=True,
    )
    logger.info("Added job 'delete_completed_constructions'.")

    scheduler.add_job(
        delete_confirmed_voc,
        trigger=CronTrigger(second="*/30"),
        id="my_job_b",
        max_instances=1,
        replace_existing=True,
    )
    logger.info("Added job 'delete_confirmed_voc'.")

    try:
        logger.info("Starting scheduler...")
        scheduler.start()
    except KeyboardInterrupt:
        logger.info("Stopping scheduler...")
        scheduler.shutdown()
        logger.info("Scheduler shut down successfully!")

 

---

Django 애플리케이션이 로드될 때 실행되는 ready() 메서드를 app에 포함시켰다.

스케줄러를 시작해야하는지 확인하고, 자동으로 스케줄러가 시작된다.

 

[Gongsa/appas.py]

from django.apps import AppConfig
from django.conf import settings

class GongsaConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "Gongsa"

    def ready(self):
        if settings.SCHEDULER_DEFAULT:
            from . import runapscheduler
            runapscheduler.start()