آشنایی با کتابخانه schedule در پایتون

کتابخانه schedule یک ابزار سبک و انسان‌پسند برای زمانبندی وظایف در پایتون است. این کتابخانه امکان تعریف تسک‌های تکرارشونده (مثلاً هر ۵ دقیقه یا هر روز ساعت ۰۸:۰۰) را با سینتکس خوانا و ساده فراهم می‌کند. بر خلاف کرون که مبتنی بر رشته‌های متنی است، schedule با توابع و اشیاء کار می‌کند و مناسب اسکریپت‌های کوچک تا متوسط است.

ویژگی‌های اصلی

• سینتکس ساده: every().minute.do(func)
• حمایت از واحدهای زمانی رایج: seconds, minutes, hours, days, weeks
• امکان تعیین روزهای هفته و زمان مشخص: every().monday.at(“10:30”)
• قابلیت تگ‌گذاری و لغو گروهی تسک‌ها
• قابلیت مشاهده next_run برای هر Job

محدودیت‌های مهم

• خودِ schedule حلقه اجرایی دائمی ندارد؛ باید با run_pending() آن را در حلقه یا ترد اجرا کنید.
• پشتیبانی از timezone داخلی ندارد؛ زمان‌ها بر اساس زمان محلی سیستم هستند.
• برای کارهای بسیار پیچیده، تراکنش‌محور یا با نیاز به تضمین اجرا یک‌بار، بهتر است از ابزارهای قوی‌تری مثل Celery یا APScheduler استفاده کنید.

نمونه‌های پایه و کاربردی

نمونه ساده: اجرای تابع هر ۱۰ ثانیه

import schedule
import time

def job():
    print("Hello, world!")

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

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

در این مثال یک تابع ساده تعریف شده که هر ۱۰ ثانیه اجرا می‌شود. تابع run_pending() وظیفه بررسی و اجرای تسک‌های زمان‌بندی‌شده را بر عهده دارد و معمولاً داخل یک حلقهٔ نامحدود قرار می‌گیرد. افزودن time.sleep(1) باعث مصرف کمتر CPU می‌شود.

اجرای روزانه در ساعت مشخص

import schedule
import time
from datetime import datetime

def backup():
    print("Running backup at", datetime.now())

schedule.every().day.at("02:30").do(backup)

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

با every().day.at("HH:MM") می‌توان تسک‌های روزانه را در ساعت مشخص اجرا کرد. توجه کنید که فرمت ساعت باید “HH:MM” باشد و بر اساس زمان محلی سیستم عمل می‌کند.

API کلیدی: جدول مرجع

اجرای schedule در پس‌زمینه با Thread

import schedule
import time
import threading

def job():
    print("Background job")

schedule.every(1).minutes.do(job)

stop_event = threading.Event()

def run_scheduler():
    while not stop_event.is_set():
        schedule.run_pending()
        time.sleep(1)

t = threading.Thread(target=run_scheduler, daemon=True)
t.start()

# برنامهٔ اصلی ادامه پیدا می‌کند
try:
    while True:
        time.sleep(10)
except KeyboardInterrupt:
    stop_event.set()
    t.join()

این الگو به شما اجازه می‌دهد تا زمانبندی در یک ترد جداگانه اجرا شود و برنامهٔ اصلی کارهای دیگری انجام دهد. از Event برای متوقف‌سازی ایمن استفاده شده است. همچنین ترد با daemon=True می‌تواند با خروج برنامه بسته شود؛ اما برای پاکسازی بهتر از Event و join استفاده می‌کنیم.

ادغام با asyncio

کتابخانه schedule به‌صورت بُردی async نیست، اما می‌توان آن را با asyncio ترکیب کرد. نمونه زیر اجرای run_pending() را در حلقهٔ asyncio زمان‌بندی می‌کند:

import schedule
import asyncio

def job():
    print("Async-friendly job")

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

async def scheduler_loop():
    while True:
        schedule.run_pending()
        await asyncio.sleep(1)

async def main():
    task = asyncio.create_task(scheduler_loop())
    await asyncio.sleep(20)  # شبیه‌سازی کار اصلی
    task.cancel()

asyncio.run(main())

در این مثال از asyncio.sleep استفاده شده تا حلقهٔ asyncio مسدود نشود. اگر وظایف شما خودشان async هستند، می‌توانید درون یک تابع wrapper از asyncio.create_task برای اجرای تابع غیرهمزمان استفاده کنید.

مدیریت خطاها و تکرار اجرا

• هر تابعی که به do() داده می‌شود باید استحکام لازم را داشته باشد یا درون try/except باشد تا خطاها باعث توقف scheduler نشوند.
• برای جلوگیری از اجرای همزمان یک تسک (race) در تردهای متعدد، از قفل (Lock) استفاده کنید یا طراحی تسک را idempotent کنید.
• وقتی نیاز به اجرای یک بار (one-shot) دارید، می‌توانید پس از اجرا job را لغو کنید یا شرطی در تابع قرار دهید.

نمونه: اجرای یک بار و حذف خودکار

import schedule
import time

def one_shot():
    print("Run once")
    return schedule.CancelJob

schedule.every().day.at("23:59").do(one_shot)

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

اگر تابع مقدار schedule.CancelJob را برگرداند، آن Job پس از اجرا حذف می‌شود. این روش برای اجرای یک بار مناسب است.

نکات پیشرفته و بهترین شیوه‌ها

• برای مدیریت timezone از کتابخانه‌هایی مثل pytz یا pendulum استفاده کنید و زمان‌های محلی را قبل از ارسال به schedule تنظیم کنید.
• برای بارگذاری/ذخیرهٔ تنظیمات (persistence) jobها باید خودتان وضعیت را ذخیره کنید؛ schedule این قابلیت را ندارد.
• برای بارگذاری متغیرهای پیکربندی (مثلاً بازه‌ها یا زمان‌ها) از فایل YAML یا دیتابیس استفاده کنید تا تغییر زمانبندی بدون تغییر کد ممکن شود.
• اگر نیاز به سرور زمانبندی توزیع‌شده یا تضمین اجرای once-only دارید، از سیستم‌هایی مانند Celery، APScheduler یا Airflow بهره ببرید.

جمع‌بندی و توصیه‌ها

کتابخانه schedule ابزار ساده و مناسبی برای بسیاری از اسکریپت‌ها و سرویس‌های سبک است که نیاز به اجرای دوره‌ای وظایف دارند. برای استفاده بهینه:

• وظایف را کوچک، مقاوم و idempotent بنویسید.
• اجرا در پس‌زمینه را با Thread یا asyncio مدیریت کنید.
• برای نیازهای پیچیده‌تر یا توزیع‌شده به ابزارهای حرفه‌ای‌تر فکر کنید.

در صورتی که مایل باشید، می‌توانم نمونه‌های عملی بیشتری شامل حفظ وضعیت jobها، مدیریت خطاهای شبکه در تسک‌ها و الگوهای طراحی برای زمانبندی در سرویس‌های واقعی ارائه دهم.

لطفاً از کمبود ها و مشکلات این محتوا برای ما بنویسید

ارسال