pkglog
EN

Python 작업 스케줄링: schedule·APScheduler·cron으로 자동화하기

작성 · · 16분 읽기 · 수정 중급 튜토리얼

이 글의 핵심

while True 루프로 돌리던 스크립트는 프로세스가 죽으면 조용히 멈추고, 터미널에서 잘 돌던 코드가 cron에서는 경로와 환경변수 차이로 실패하곤 합니다. 라이브러리 스케줄러와 OS 스케줄러 중 무엇을 고를지 정하고, 로깅·예외 처리·타임아웃으로 무인 작업을 안정적으로 운영하는 방법을 정리합니다.

들어가며

배치 스크립트나 서버 안에서 주기적으로 작업을 돌리려면, 단순 while 루프보다 스케줄 라이브러리나 OS 스케줄러와 맞추는 편이 안전합니다. 작업 스케줄링은 정해진 시간에 자동으로 작업을 실행하는 기술입니다.

선택지는 크게 두 갈래입니다. 하나는 Python 프로세스를 계속 띄워 두고 그 안에서 시각을 확인하는 라이브러리 스케줄러(schedule, APScheduler)이고, 다른 하나는 OS가 정해진 시각에 스크립트를 새로 실행해 주는 OS 스케줄러(cron, Windows 작업 스케줄러)입니다. 라이브러리 방식은 코드 안에서 일정을 관리하고 메모리 상태를 공유하기 쉬운 대신, 프로세스가 죽으면 이후 작업이 모두 조용히 멈춥니다. OS 방식은 매번 새 프로세스라 한 번 실패해도 다음 실행에 영향이 없고 서버 재부팅 후에도 계속 동작하지만, 실행 환경이 터미널과 달라 처음 설정할 때 시행착오가 있습니다. 하루 몇 번 도는 독립적인 작업이라면 OS 스케줄러가, 웹 서버 안에서 몇 분마다 캐시를 갱신하는 식이라면 라이브러리 스케줄러가 대체로 잘 맞습니다.

schedule 라이브러리

설치

pip install schedule

기본 사용

import schedule
import time
def job():
    print("작업 실행!")
# 10초마다
schedule.every(10).seconds.do(job)
# 1분마다
schedule.every(1).minutes.do(job)
# 매일 오전 9시
schedule.every().day.at("09:00").do(job)
# 매주 월요일 오전 10시
schedule.every().monday.at("10:00").do(job)
# 실행
while True:
    schedule.run_pending()
    time.sleep(1)

schedule은 스스로 시간을 재지 않습니다. run_pending()이 호출될 때마다 “실행 시각이 지난 작업이 있는가”를 확인해 실행할 뿐이라, 마지막의 while 루프가 없으면 아무 일도 일어나지 않습니다. time.sleep(1)은 1초마다 확인한다는 뜻이고, 이 값이 크면 그만큼 실행이 늦어질 수 있습니다.

이 구조에서 알아 둘 동작이 몇 가지 있습니다. 첫째, 모든 작업은 메인 스레드에서 하나씩 순서대로 실행됩니다. 10초마다 도는 작업이 30초 걸리면 그동안 다른 작업은 모두 밀립니다. 둘째, 작업 안에서 예외가 나면 run_pending()까지 전파되어 스케줄러 루프 전체가 종료됩니다. 한 번의 네트워크 오류로 밤새 돌아야 할 스크립트가 멈추는 흔한 원인이라, 작업 함수는 마지막 절처럼 예외를 잡아 기록하도록 감싸야 합니다. 셋째, 프로세스가 꺼져 있던 동안의 실행은 나중에 보충되지 않습니다. at("09:00")은 서버 시스템의 로컬 시간 기준이라, UTC로 설정된 클라우드 서버에서는 한국 시간 오후 6시에 실행됩니다. 최근 버전은 at("09:00", "Asia/Seoul")처럼 시간대를 지정할 수 있습니다(pytz 필요).

자동 백업 스크립트

import schedule
import time
from datetime import datetime
import shutil
from pathlib import Path
def backup_files():
    """파일 백업"""
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_name = f"backup_{timestamp}"
    
    source = Path('./data')
    backup_dir = Path('./backups')
    backup_dir.mkdir(exist_ok=True)
    
    backup_path = backup_dir / backup_name
    shutil.copytree(source, backup_path)
    
    print(f"[{datetime.now()}] 백업 완료: {backup_name}")
# 매일 자정에 백업
schedule.every().day.at("00:00").do(backup_files)
# 매주 일요일 오후 11시에 백업
schedule.every().sunday.at("23:00").do(backup_files)
print("백업 스케줄러 시작...")
while True:
    schedule.run_pending()
    time.sleep(60)

타임스탬프를 폴더 이름에 넣어 백업마다 새 폴더가 생기게 했습니다. copytree는 대상 폴더가 이미 있으면 FileExistsError를 내므로 같은 이름으로 덮어쓰는 사고는 막아 주지만, 일요일 밤처럼 두 작업이 가까운 시각에 겹치지 않는지는 확인해 볼 만합니다. 또 이 함수는 오래된 백업을 지우지 않아서 매일 원본 크기만큼 디스크가 늘어납니다. 파일 자동화의 cleanup_old_backups처럼 최근 N개만 남기는 단계를 함께 두어야 몇 달 뒤 디스크가 가득 차는 일을 피할 수 있습니다.

time.sleep(60)으로 1분마다 확인하므로 실행 시각이 최대 1분 가까이 늦어질 수 있습니다. 백업처럼 정확한 초가 중요하지 않은 작업에는 CPU를 덜 쓰는 합리적인 선택입니다. 상대 경로 ./data는 스크립트 위치가 아니라 실행한 위치 기준이라, 다른 폴더에서 실행하거나 서비스로 등록하면 엉뚱한 곳을 백업하거나 FileNotFoundError가 납니다. Path(__file__).resolve().parent / 'data'처럼 스크립트 위치를 기준으로 경로를 만드는 편이 안전합니다.

APScheduler

설치

pip install apscheduler

고급 스케줄링

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
scheduler = BlockingScheduler()
def job1():
    print(f"[{datetime.now()}] Job 1 실행")
def job2():
    print(f"[{datetime.now()}] Job 2 실행")
# Cron 스타일
scheduler.add_job(job1, 'cron', hour=9, minute=0)  # 매일 9시
# 간격
scheduler.add_job(job2, 'interval', minutes=30)  # 30분마다
# 특정 시간
scheduler.add_job(
    job1,
    'cron',
    day_of_week='mon-fri',
    hour=9,
    minute=0
)
scheduler.start()

APScheduler는 “언제 실행할지”를 트리거로 표현합니다. 'cron'은 cron과 같은 필드(day_of_week, hour, minute 등)로 달력 기준 시각을, 'interval'은 시작 시점부터의 고정 간격을, 'date'는 한 번만 실행할 특정 시각을 지정합니다. CronTrigger.from_crontab('0 9 * * mon-fri')로 cron 문자열을 그대로 쓸 수도 있습니다. 예제에서 job1은 두 번 등록되어 평일 9시에는 두 번 실행된다는 점도 확인해 두세요.

BlockingScheduler는 start()에서 멈춰 스케줄러만 도는 스크립트용이고, 웹 서버처럼 다른 일을 하는 프로그램 안에서는 BackgroundScheduler를 써서 별도 스레드로 돌립니다. schedule과 가장 큰 차이는 작업이 스레드 풀에서 실행된다는 점으로, 오래 걸리는 작업이 다른 작업을 막지 않습니다. 대신 같은 작업이 이전 실행이 끝나기 전에 다시 시작될 수 있는데, 기본값(max_instances=1)이면 이 경우 “maximum number of running instances reached” 경고와 함께 새 실행을 건너뜁니다. 서버가 잠시 멈춰 실행 시각을 놓쳤을 때 허용할 지연은 misfire_grace_time, 여러 번 놓친 실행을 한 번으로 합칠지는 coalesce로 정합니다.

주의할 점은 웹 서버에 BackgroundScheduler를 넣으면 Gunicorn 워커 수만큼 스케줄러가 생겨 같은 작업이 워커 수만큼 중복 실행된다는 것입니다. 개발 서버에서는 한 번 돌다가 배포 후 이메일이 네 통씩 발송되는 식으로 드러납니다. 스케줄러는 별도 프로세스 하나로 분리하거나, 작업 저장소를 DB로 두고 잠금을 거는 방식으로 해결합니다. 참고로 pip install apscheduler는 현재 3.x 안정 버전을 설치하며, 4.x는 API가 크게 달라 이 예제가 그대로 동작하지 않으므로 버전을 고정해 두는 것이 좋습니다.

웹 스크래핑 자동화

주기적 데이터 수집

import schedule
import time
import requests
from bs4 import BeautifulSoup
import pandas as pd
from datetime import datetime
def scrape_prices():
    """상품 가격 수집"""
    urls = [
        'https://shop1.com/product/123',
        'https://shop2.com/product/456'
    ]
    
    prices = []
    
    for url in urls:
        try:
            response = requests.get(url, timeout=10)
            soup = BeautifulSoup(response.text, 'html.parser')
            
            price = soup.select_one('.price').text
            price = int(price.replace(',', '').replace('원', ''))
            
            prices.append({
                'url': url,
                'price': price,
                'timestamp': datetime.now()
            })
        except Exception as e:
            print(f"에러: {url} - {e}")
    
    # CSV에 추가
    df = pd.DataFrame(prices)
    df.to_csv('price_history.csv', mode='a', header=False, index=False)
    print(f"[{datetime.now()}] 가격 수집 완료")
# 1시간마다 실행
schedule.every(1).hours.do(scrape_prices)
while True:
    schedule.run_pending()
    time.sleep(60)

URL마다 try로 감싼 이유는 한 사이트가 응답하지 않거나 페이지 구조가 바뀌어도 나머지 상품의 가격은 수집하기 위해서입니다. timeout=10도 중요합니다. requests.get은 타임아웃을 주지 않으면 서버가 응답하지 않을 때 무한정 기다리고, schedule은 작업을 순서대로 실행하므로 이후 모든 수집이 멈춥니다. 무인 스크립트가 “죽지는 않았는데 아무 일도 안 하는” 상태가 되는 가장 흔한 원인입니다.

soup.select_one('.price')는 요소를 찾지 못하면 None을 돌려주고, 이어지는 .text에서 AttributeError가 납니다. 사이트 개편으로 클래스 이름이 바뀌면 매시간 같은 에러가 쌓이므로, 로그를 주기적으로 확인하거나 연속 실패 시 알림을 보내는 장치가 있어야 합니다. response.raise_for_status()를 호출해 404·500 응답도 실패로 처리하는 편이 좋습니다. 수집 주기를 짧게 잡으면 대상 사이트에 부담을 주고 차단될 수 있으니, robots.txt와 이용 약관을 확인하고 필요한 만큼만 요청하세요. 스크래핑 자체는 웹 스크래핑에서 다룹니다.

CSV에 mode='a', header=False로 이어 붙이면 파일이 처음 만들어질 때도 헤더가 없어서 나중에 read_csv로 읽으면 첫 데이터 행이 열 이름이 됩니다. header=not Path('price_history.csv').exists()처럼 파일이 없을 때만 헤더를 쓰게 하면 해결됩니다. 모든 URL이 실패해 prices가 비어 있으면 빈 DataFrame이 저장되는 것도 확인해 둘 부분입니다.

이메일 알림

자동 리포트 발송

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime
import schedule
def send_daily_report():
    """일일 리포트 이메일 발송"""
    # 리포트 생성
    report = generate_report()
    
    # 이메일 설정
    msg = MIMEMultipart()
    msg['From'] = '[email protected]'
    msg['To'] = '[email protected]'
    msg['Subject'] = f"일일 리포트 - {datetime.now().strftime('%Y-%m-%d')}"
    
    msg.attach(MIMEText(report, 'html'))
    
    # 발송
    try:
        server = smtplib.SMTP('smtp.gmail.com', 587)
        server.starttls()
        server.login('[email protected]', 'password')
        server.send_message(msg)
        server.quit()
        print("리포트 발송 완료")
    except Exception as e:
        print(f"발송 실패: {e}")
def generate_report():
    """리포트 HTML 생성"""
    return """
    <html>
        <body>
            <h1>일일 리포트</h1>
            <p>총 매출: 1,000,000원</p>
            <p>신규 고객: 50명</p>
        </body>
    </html>
    """
# 매일 오전 8시에 발송
schedule.every().day.at("08:00").do(send_daily_report)

MIMEMultipart로 메일을 만들고 HTML 본문을 첨부한 뒤 SMTP 서버로 보냅니다. 587번 포트는 평문으로 연결한 다음 starttls()로 암호화를 시작하는 방식이고, 465번 포트를 쓰는 서버라면 처음부터 암호화된 smtplib.SMTP_SSL을 써야 합니다. 이 스크립트도 스케줄 등록만 하고 run_pending() 루프가 없으므로, 앞 예제처럼 루프를 붙이거나 OS 스케줄러로 send_daily_report()를 한 번 실행하는 스크립트로 만들어야 합니다.

Gmail은 일반 계정 비밀번호로 SMTP 로그인을 허용하지 않아서 예제를 그대로 실행하면 SMTPAuthenticationError: (535, ... Username and Password not accepted)가 납니다. 2단계 인증을 켠 뒤 발급받은 앱 비밀번호를 써야 하고, 회사 메일이라면 관리자가 SMTP 접근을 막아 두었을 수 있습니다. 비밀번호는 코드에 적지 말고 환경 변수에서 읽으세요. 그런데 cron으로 실행하면 셸의 환경 변수를 읽지 못하므로 crontab에 변수를 직접 정의하거나 .env 파일을 읽는 방식을 써야 합니다.

server.quit()이 try 안에 있어 send_message에서 예외가 나면 연결이 닫히지 않습니다. with smtplib.SMTP(...) as server: 형태로 쓰면 예외가 나도 연결이 정리됩니다. 발송 실패를 print로만 남기면 무인 실행에서는 아무도 보지 못하므로, 로그 파일에 기록하고 실패가 반복되면 다른 채널로 알리는 편이 좋습니다.

백그라운드 실행

Windows 작업 스케줄러

# Python 스크립트를 Windows 작업 스케줄러에 등록
# 1. 작업 스케줄러 열기
# 2. 기본 작업 만들기
# 3. 프로그램: python.exe
# 4. 인수: C:\path\to\script.py

작업 스케줄러에서 가장 자주 놓치는 항목은 “시작 위치(Start in)“입니다. 비워 두면 작업 디렉터리가 C:\Windows\System32 같은 곳이 되어 스크립트 안의 상대 경로 파일을 찾지 못합니다. 스크립트가 있는 폴더를 시작 위치로 지정하세요. “프로그램”에는 python.exe 대신 가상 환경의 C:\path\to\.venv\Scripts\python.exe를 절대 경로로 적어야 설치한 패키지를 찾을 수 있고, 콘솔 창이 뜨지 않게 하려면 pythonw.exe를 씁니다. 로그인하지 않은 상태에서도 실행하려면 “사용자의 로그온 여부에 관계없이 실행”을 선택해야 하며, 이때는 네트워크 드라이브 매핑 같은 사용자 세션 자원을 쓸 수 없습니다.

Linux cron

# crontab 편집
crontab -e
# 매일 오전 9시
0 9 * * * /usr/bin/python3 /path/to/script.py
# 매시간
0 * * * * /usr/bin/python3 /path/to/script.py
# 매주 월요일 오전 10시
0 10 * * 1 /usr/bin/python3 /path/to/script.py

cron 표현식의 다섯 필드는 순서대로 분(0-59), 시(0-23), 일(1-31), 월(1-12), 요일(0-7, 0과 7이 일요일)입니다. */15 * * * *는 15분마다, 0 9 * * 1-5는 평일 오전 9시입니다. 필드 순서를 헷갈려 9 0 * * *(매일 0시 9분)로 쓰는 실수가 흔하므로, 등록 전에 crontab.guru 같은 도구로 해석을 확인하면 좋습니다.

cron에 처음 등록한 스크립트가 “아무 반응이 없는” 경우는 대부분 출력과 오류를 볼 방법이 없어서입니다. 0 9 * * * /path/.venv/bin/python /path/to/script.py >> /path/to/cron.log 2>&1처럼 표준 출력과 표준 오류를 로그 파일로 보내 두면 ModuleNotFoundError나 FileNotFoundError 같은 원인이 바로 보입니다. 가상 환경을 쓴다면 /usr/bin/python3 대신 가상 환경 안의 python을 절대 경로로 적어야 설치한 패키지가 보입니다. 또 cron 명령줄에서 %는 줄바꿈으로 해석되므로 date +%Y%m%d 같은 명령을 넣으려면 \%로 이스케이프해야 합니다. 서버 시간대가 UTC라면 한국 시간 기준 일정은 9시간을 빼서 적어야 한다는 점도 잊기 쉽습니다.

작업이 다음 실행 시각까지 끝나지 않으면 cron은 이전 실행을 기다리지 않고 새 프로세스를 또 띄웁니다. 5분마다 도는 작업이 가끔 10분 걸리면 두 개가 동시에 같은 파일을 쓰게 되므로, flock -n /tmp/job.lock python script.py처럼 잠금을 걸어 중복 실행을 막는 방법이 있습니다.

로깅·예외·타임아웃으로 스케줄 작업 운영하기

while True 안에서 run_pending을 도는 방식은 알람을 맞춰 두고 정해진 함수만 반복 호출하는 시계와 비슷합니다. 장시간 돌리려면 로그 파일으로 실행 여부를 남기고, 작업 안에서 예외를 삼켜 버리지 말고 기록하며, 오래 걸리는 작업에는 타임아웃(환경에 따라 signal 등)을 검토합니다.

# ✅ 로깅 추가
import logging
logging.basicConfig(
    filename='scheduler.log',
    level=logging.INFO,
    format='%(asctime)s - %(message)s'
)
def job():
    logging.info("작업 시작")
    # 작업 수행
    logging.info("작업 완료")
# ✅ 에러 처리
def safe_job():
    try:
        job()
    except Exception as e:
        logging.error(f"에러: {e}")
# ✅ 타임아웃 설정
import signal
def timeout_handler(signum, frame):
    raise TimeoutError("작업 시간 초과")
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(300)  # 5분 제한

safe_job처럼 작업을 감싸 두면 예외가 나도 스케줄러 루프는 계속 돕니다. 다만 logging.error(f"에러: {e}")는 메시지만 남기고 어느 줄에서 났는지는 잃어버리므로, logging.exception("작업 실패")를 쓰면 트레이스백까지 기록됩니다. 로그 파일은 계속 커지므로 logging.handlers.RotatingFileHandler나 TimedRotatingFileHandler로 크기나 날짜 기준으로 나눠 두세요. basicConfig(filename='scheduler.log')도 상대 경로라 cron에서 실행하면 예상과 다른 폴더에 로그가 생깁니다.

signal.alarm을 이용한 타임아웃에는 제약이 큽니다. SIGALRM은 Windows에 없어서 이 코드는 Windows에서 AttributeError로 실패하고, Unix에서도 메인 스레드에서만 시그널 핸들러를 등록할 수 있어 APScheduler의 작업 스레드 안에서는 쓸 수 없습니다. 또 예제처럼 모듈 최상단에서 한 번 alarm(300)을 걸면 스크립트 전체가 5분 뒤 중단되므로, 실제로는 작업 시작 시 alarm(300), 끝나면 alarm(0)으로 해제해야 합니다. 이식성이 필요하다면 네트워크 호출마다 타임아웃을 지정하는 것이 가장 확실하고, 작업 전체에 제한을 걸어야 한다면 concurrent.futures로 별도 프로세스에서 실행하고 future.result(timeout=300)으로 기다리는 방법이 있습니다. 스레드는 밖에서 강제로 멈출 수 없으므로 프로세스를 쓰는 것이 핵심입니다.

같이 보면 좋은 글

자주 묻는 질문 (FAQ)

Q. 터미널에서는 잘 돌던 스크립트가 cron에 등록하면 실행되지 않는 이유는 무엇인가요?

A. cron은 로그인 셸과 다른 최소한의 환경 변수와 작업 디렉터리로 실행되기 때문에, python3나 상대 경로 파일을 찾지 못하는 경우가 많습니다. 예제처럼 /usr/bin/python3 /path/to/script.py로 인터프리터와 스크립트를 모두 절대 경로로 적고, 스크립트 안의 파일 경로도 절대 경로로 다루는 것이 안전합니다. 실행 여부를 확인할 수 있도록 logging으로 시작·완료·에러를 파일에 남겨 두면 원인을 찾기 훨씬 쉽습니다.