Python 에서 정기적으로 함수를 실행 하는 방법

[pai-plznw4me]

python django 에서 정기적으로 함수를 실행하고 싶다.

해당 블로그를 보면 python 함수를 정기적으로 실행할 수 있는 방법은 크게 3가지가 있는데

Celery 와 Redis 의 활용
리눅스의 CronTab 설정의 활용
파이썬 APScheduler 를 활용

해당 이슈에서는 Python APScheduler 을 활용해서 scheduler 을 생성한다.

블로그에서 적용 하는 부분은 파이썬 APScheduler 를 활용 부분. 관련 blog

[pai-plznw4me]

프로젝트 이름: crawling_server

django-admin startproject crawling_server
cd crawling_server

앱 이름 : stock_crawling

python manage.py startapp stock_crawling

목적 : 'hello world' 을 매 1초 마다 출력

1. 관련 api. 설치

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.

settings.py 에 아래 코드를 추가

INSTALLED_APPS = [
    ... 
    "stock_crawling",
    'django_apscheduler',
]

APSCHEDULER_DATETIME_FORMAT = "N j, Y, f:s a"  # Default

SCHEDULER_DEFAULT = True

3.

view 파일에 아래 코드 작성

def send_hello():
    print('hello world')

4

project 폴더에 operator.py script 생성

./
├── crawling_server <- **project dir** 
│   ├── asgi.py
│   ├── operator.py <- **add 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

operator.py script 에 아래 코드 붙여 넣기

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()

scheduler 을 작동시키는 trigger 에는 cron 과 interval 이 대표적인데 cron 을 실행하고 싶으면 decorator 부분을 변경

@scheduler.scheduled_job('interval', seconds=1, name='auto_hello')

example cron trigger

@scheduler.scheduled_job('cron', hour=1, name='auto_hello')

5.

project dir 내 apps.py script 에 아래 코드 추가

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.

python manage.py runserver --noreload

7.

결과

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]