Python 정기 작업 스케줄링

Python 정기 작업 스케줄링

정기적으로 반복되어야 하는 작업을 처리하는 스케줄링은 다양한 분야에서 유용하게 활용됩니다. 예를 들어, 로그 파일 정리, 데이터 백업, 주기적인 알림 전송 등이 있습니다. Python에서는 이러한 정기 작업 스케줄링을 구현할 수 있는 여러 가지 강력한 도구를 제공합니다. 이번 포스팅에서는 schedule 라이브러리와 Python의 표준 라이브러리인 threading 및 datetime을 활용하여 정기 작업을 스케줄링하는 방법을 다룹니다.

1. Python에서의 작업 스케줄링

Python은 작업 스케줄링을 위해 다양한 라이브러리를 제공합니다. 가장 많이 사용되는 라이브러리로는 다음과 같은 것들이 있습니다:

• schedule: 간단한 정기 작업 스케줄링 라이브러리로 사용법이 매우 직관적입니다.
• APScheduler: 고급 스케줄링 기능을 제공하며, 데이터베이스 백엔드 지원과 복잡한 일정 설정이 가능합니다.
• celery: 분산 작업 처리를 지원하는 강력한 툴입니다.

이번 포스팅에서는 간단한 사용법과 직관성을 제공하는 schedule 라이브러리를 중점적으로 다룹니다.

2. schedule 라이브러리 설치

먼저, schedule 라이브러리를 설치해야 합니다. 터미널에서 다음 명령어를 실행하세요:

pip install schedule

3. 기본적인 사용법

schedule 라이브러리를 사용하면 특정 시간 간격으로 작업을 쉽게 등록할 수 있습니다. 아래는 기본적인 사용 예제입니다:

import schedule
import time

def job():
    print("작업이 실행되었습니다!")

# 1분마다 작업 실행
schedule.every(1).minutes.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

코드 설명

1. schedule.every(1).minutes.do(job): 1분마다 job 함수를 실행하도록 스케줄링합니다.
2. schedule.run_pending(): 스케줄러에 등록된 작업 중 실행할 작업을 실행합니다.
3. time.sleep(1): CPU 사용량을 줄이기 위해 1초 대기합니다.

4. 다양한 작업 스케줄링 옵션

schedule 라이브러리는 다양한 스케줄링 옵션을 제공합니다:

• 매일 특정 시간에 작업 실행

schedule.every().day.at("10:30").do(job)

• 매주 특정 요일에 작업 실행

schedule.every().monday.do(job)

• 매시간 실행

schedule.every().hour.do(job)

• 초 단위로 실행 (테스트용)

schedule.every(10).seconds.do(job)

5. 스케줄링 작업에 매개변수 전달

작업 함수에 매개변수를 전달하고 싶다면 다음과 같이 작성할 수 있습니다:

def job_with_params(param):
    print(f"{param} 작업이 실행되었습니다!")

schedule.every(1).minutes.do(job_with_params, param="백업")

실행 결과

1분마다 "백업 작업이 실행되었습니다!"라는 메시지가 출력됩니다.

6. 멀티스레딩을 활용한 비동기 스케줄링

기본적으로 schedule은 단일 스레드에서 작동합니다. 하지만 멀티스레딩을 사용하면 동시에 여러 작업을 실행할 수 있습니다.

import threading

def threaded_job():
    print("스레드에서 실행 중인 작업")

# 스레드로 작업 실행
schedule.every(10).seconds.do(lambda: threading.Thread(target=threaded_job).start())

while True:
    schedule.run_pending()
    time.sleep(1)

7. 실제 사례: 정기적인 로그 파일 정리

아래는 정기적으로 로그 파일을 정리하는 예제입니다:

import os
from datetime import datetime

def clean_logs():
    log_directory = "./logs"
    now = datetime.now()
    for filename in os.listdir(log_directory):
        file_path = os.path.join(log_directory, filename)
        if os.path.isfile(file_path):
            file_creation_time = datetime.fromtimestamp(os.path.getctime(file_path))
            if (now - file_creation_time).days > 7:  # 7일 이상 지난 파일 삭제
                os.remove(file_path)
                print(f"삭제된 파일: {filename}")

schedule.every().day.at("02:00").do(clean_logs)

while True:
    schedule.run_pending()
    time.sleep(1)

코드 설명

1. os.listdir와 os.path.getctime을 사용해 파일 생성 시간을 가져옵니다.
2. 생성된 지 7일 이상 된 파일을 삭제합니다.
3. 매일 새벽 2시에 작업이 실행됩니다.

8. 종료 시그널 처리

작업 실행 중 Ctrl+C로 프로그램을 종료하려고 할 때, 추가 작업을 수행하려면 다음과 같이 처리할 수 있습니다:

import signal
import sys

def cleanup_before_exit(signal, frame):
    print("프로그램 종료 중...")
    sys.exit(0)

signal.signal(signal.SIGINT, cleanup_before_exit)

while True:
    schedule.run_pending()
    time.sleep(1)

결론

Python의 schedule 라이브러리는 간단하고 직관적인 인터페이스로 정기 작업 스케줄링을 손쉽게 구현할 수 있게 해줍니다. 이 포스팅에서 소개한 기본 사용법과 실전 예제를 활용하여 다양한 작업을 자동화하고 시간을 절약하세요! 앞으로 더 복잡한 스케줄링 요구사항이 있을 경우 APScheduler와 같은 고급 도구를 검토해보는 것도 추천드립니다.