Python ile Cron Alternatifi: schedule Kütüphanesi Kullanımı

Cron iyi bir araçtır, bunu kimse inkar edemez. Yıllardır sunucularımızın vazgeçilmezi oldu. Ama şunu söylemek lazım: Cron syntax’ını her seferinde araştırmak zorunda kalmak, hata mesajlarının neredeyse hiç anlamlı olmaması ve karmaşık zamanlama mantığını tek satıra sıkıştırmaya çalışmak bazen insanı çıldırtıyor. İşte tam bu noktada Python’un schedule kütüphanesi devreye giriyor. Eğer zaten Python ile script yazıyorsanız, zamanlama mantığını da Python içinde tutmak hem okunabilirlik hem de bakım kolaylığı açısından ciddi avantaj sağlıyor.

schedule Kütüphanesi Nedir?

schedule, Python ile yazılmış hafif bir görev zamanlama kütüphanesidir. Bir daemon veya arka plan servisi değildir, sadece Python kodunuzun içinde belirli aralıklarla fonksiyon çağırmanızı sağlayan sade bir araçtır. 2013 yılında Daniel Bader tarafından geliştirilmiş olan bu kütüphane, insan tarafından okunabilir bir API sunmasıyla öne çıkıyor.

Cron’un /5 * gibi kriptik syntax’ı yerine şunu yazıyorsunuz:

schedule.every(5).minutes.do(benim_fonksiyonum)

Bu kadar. Başka bir şey öğrenmenize gerek yok.

Kurulum

pip install schedule
# veya requirements.txt'e ekleyip
pip install -r requirements.txt
# Belirli bir versiyon için
pip install schedule==1.2.1

Kütüphane oldukça küçük, bağımlılıkları minimal. Üretim ortamında sanal ortam kullanmayı unutmayın:

python3 -m venv /opt/myproject/venv
source /opt/myproject/venv/bin/activate
pip install schedule

Temel Kullanım ve Syntax

İlk örneğimize bakalım. En basit haliyle bir zamanlama scripti şöyle görünür:

import schedule
import time
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def disk_kullanimi_kontrol():
    import shutil
    toplam, kullanilan, bos = shutil.disk_usage("/")
    yuzde = (kullanilan / toplam) * 100
    logging.info(f"Disk kullanimi: %{yuzde:.1f}")
    if yuzde > 85:
        logging.warning(f"UYARI: Disk dolulugu %{yuzde:.1f} seviyesine ulasti!")

def log_temizle():
    import subprocess
    subprocess.run(["find", "/var/log/myapp", "-name", "*.log", 
                   "-mtime", "+30", "-delete"])
    logging.info("Eski loglar temizlendi.")

# Her 10 dakikada bir disk kontrolü
schedule.every(10).minutes.do(disk_kullanimi_kontrol)

# Her gün saat 02:00'de log temizleme
schedule.every().day.at("02:00").do(log_temizle)

# Her Pazartesi sabah rapor gönder
schedule.every().monday.at("09:00").do(lambda: logging.info("Haftalik rapor gonderildi."))

logging.info("Zamanlayici basladi.")

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

Buradaki run_pending() çağrısı, zamanı gelen görevleri çalıştırıyor. time.sleep(1) ise CPU’yu gereksiz yere meşgul etmemek için gerekli. Döngü her saniye bir kez kontrol ediyor.

Zamanlama Seçenekleri

schedule kütüphanesinin sunduğu zamanlama seçeneklerini listeleyelim:

• every().second: Her saniye
• every(5).seconds: Her 5 saniyede bir
• every().minute: Her dakika
• every(30).minutes: Her 30 dakikada bir
• every().hour: Her saat
• every(2).hours: Her 2 saatte bir
• every().day.at(“HH:MM”): Her gün belirli saatte
• every().monday: Her Pazartesi (pazartesi, tuesday, wednesday, thursday, friday, saturday, sunday)
• every().week: Her hafta
• every(2).weeks: Her 2 haftada bir

Gerçek Dünya Senaryosu 1: Sistem Sağlık Monitörü

Pek çok sysadmin için en yaygın otomasyon ihtiyacı sistem kaynaklarını izlemektir. İşte kapsamlı bir örnek:

import schedule
import time
import logging
import smtplib
import psutil
import subprocess
from email.mime.text import MIMEText
from datetime import datetime

logging.basicConfig(
    filename='/var/log/sistem_monitor.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

MAIL_GONDEREN = "[email protected]"
MAIL_ALICI = "[email protected]"
SMTP_SUNUCU = "localhost"

def mail_gonder(konu, icerik):
    msg = MIMEText(icerik)
    msg['Subject'] = konu
    msg['From'] = MAIL_GONDEREN
    msg['To'] = MAIL_ALICI
    try:
        with smtplib.SMTP(SMTP_SUNUCU) as smtp:
            smtp.send_message(msg)
        logging.info(f"Mail gonderildi: {konu}")
    except Exception as e:
        logging.error(f"Mail gonderilemedi: {e}")

def cpu_kontrol():
    cpu_yuzde = psutil.cpu_percent(interval=2)
    logging.info(f"CPU kullanimi: %{cpu_yuzde}")
    if cpu_yuzde > 90:
        mail_gonder(
            "KRITIK: Yuksek CPU Kullanimi",
            f"Sunucu CPU kullanimi %{cpu_yuzde} seviyesine ulasti!nZaman: {datetime.now()}"
        )

def ram_kontrol():
    ram = psutil.virtual_memory()
    yuzde = ram.percent
    logging.info(f"RAM kullanimi: %{yuzde}")
    if yuzde > 85:
        mail_gonder(
            "UYARI: Yuksek RAM Kullanimi",
            f"RAM kullanimi %{yuzde}. Toplam: {ram.total // (1024**3)}GB, "
            f"Kullanilan: {ram.used // (1024**3)}GB"
        )

def disk_kontrol():
    for partition in psutil.disk_partitions():
        try:
            kullanim = psutil.disk_usage(partition.mountpoint)
            yuzde = kullanim.percent
            logging.info(f"Disk {partition.mountpoint}: %{yuzde}")
            if yuzde > 80:
                mail_gonder(
                    f"UYARI: Disk Dolulugu - {partition.mountpoint}",
                    f"{partition.mountpoint} bolumu %{yuzde} dolu!n"
                    f"Toplam: {kullanim.total // (1024**3)}GB"
                )
        except PermissionError:
            pass

def servis_kontrol():
    kritik_servisler = ["nginx", "postgresql", "redis"]
    for servis in kritik_servisler:
        result = subprocess.run(
            ["systemctl", "is-active", servis],
            capture_output=True, text=True
        )
        if result.stdout.strip() != "active":
            logging.error(f"KRITIK: {servis} servisi calısmiyor!")
            mail_gonder(
                f"KRITIK: {servis} Servisi Durdu",
                f"{servis} servisi aktif degil. Durum: {result.stdout.strip()}n"
                f"Zaman: {datetime.now()}"
            )

# Görev zamanlamaları
schedule.every(5).minutes.do(cpu_kontrol)
schedule.every(5).minutes.do(ram_kontrol)
schedule.every(15).minutes.do(disk_kontrol)
schedule.every(2).minutes.do(servis_kontrol)

if __name__ == "__main__":
    logging.info("Sistem monitoru basladi.")
    # Başlangıçta bir kez hepsini çalıştır
    cpu_kontrol()
    ram_kontrol()
    disk_kontrol()
    servis_kontrol()

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

Gerçek Dünya Senaryosu 2: Veritabanı Yedekleme Otomasyonu

Veritabanı yedeklemesi sysadmin’in klasik görevlerinden biri. Cron ile halledebilirsiniz elbette, ama şöyle bir senaryo düşünün: Yedeği aldıktan sonra başka bir işlem yapılacak, hata durumunda farklı bir akış izlenecek ve tüm bu mantık Python’da zaten yazılı. Bu durumda schedule çok daha temiz bir çözüm sunuyor:

import schedule
import time
import subprocess
import os
import logging
import gzip
import shutil
from datetime import datetime, timedelta
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/yedekleme.log'),
        logging.StreamHandler()
    ]
)

YEDEK_DIZIN = Path("/backup/postgresql")
DB_ADI = "uretim_db"
DB_KULLANICI = "backup_user"
YEDEK_SAKLA_GUN = 30

def yedek_al():
    tarih = datetime.now().strftime("%Y%m%d_%H%M%S")
    yedek_dosya = YEDEK_DIZIN / f"{DB_ADI}_{tarih}.sql"
    sikistir_dosya = Path(str(yedek_dosya) + ".gz")

    YEDEK_DIZIN.mkdir(parents=True, exist_ok=True)

    logging.info(f"Yedekleme baslıyor: {DB_ADI}")

    try:
        # pg_dump ile yedek al
        result = subprocess.run(
            ["pg_dump", "-U", DB_KULLANICI, "-h", "localhost", DB_ADI],
            capture_output=True,
            text=False  # binary mod, gzip için
        )

        if result.returncode != 0:
            logging.error(f"pg_dump hata verdi: {result.stderr.decode()}")
            return False

        # Sıkıştır
        with gzip.open(sikistir_dosya, 'wb') as f:
            f.write(result.stdout)

        boyut_mb = sikistir_dosya.stat().st_size / (1024 * 1024)
        logging.info(f"Yedek tamamlandi: {sikistir_dosya} ({boyut_mb:.1f}MB)")
        return True

    except Exception as e:
        logging.error(f"Yedekleme hatasi: {e}")
        return False

def eski_yedekleri_temizle():
    sinir_tarih = datetime.now() - timedelta(days=YEDEK_SAKLA_GUN)
    silinen = 0

    for dosya in YEDEK_DIZIN.glob("*.sql.gz"):
        dosya_tarihi = datetime.fromtimestamp(dosya.stat().st_mtime)
        if dosya_tarihi < sinir_tarih:
            dosya.unlink()
            silinen += 1
            logging.info(f"Eski yedek silindi: {dosya.name}")

    logging.info(f"Temizlik tamamlandi. {silinen} dosya silindi.")

def haftalik_tam_yedek():
    logging.info("Haftalik tam yedekleme baslıyor...")
    basari = yedek_al()
    if basari:
        logging.info("Haftalik tam yedekleme tamamlandi.")
    else:
        logging.error("Haftalik tam yedekleme BASARISIZ!")

# Her gece 03:00'te yedek al
schedule.every().day.at("03:00").do(yedek_al)

# Her Pazar 01:00'de eski yedekleri temizle
schedule.every().sunday.at("01:00").do(eski_yedekleri_temizle)

# Her Pazar 02:00'de haftalık tam yedek
schedule.every().sunday.at("02:00").do(haftalik_tam_yedek)

if __name__ == "__main__":
    logging.info("Yedekleme zamanlayicisi basladi.")
    while True:
        schedule.run_pending()
        time.sleep(30)  # Yedekleme için 30 saniyelik kontrol aralığı yeterli

İleri Seviye Kullanım: Thread ile Paralel Görev Çalıştırma

Varsayılan schedule kullanımında görevler sıralı çalışır. Bir görev uzun sürerse diğerleri bekler. Bunu çözmek için thread kullanabilirsiniz:

import schedule
import time
import threading
import logging
import requests

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(threadName)s - %(message)s')

def threaded_calistir(fonksiyon):
    """Verilen fonksiyonu ayrı bir thread'de çalıştırır."""
    def wrapper():
        t = threading.Thread(target=fonksiyon, name=fonksiyon.__name__)
        t.daemon = True
        t.start()
    return wrapper

def uzun_api_kontrolu():
    """Bu işlem birkaç dakika sürebilir."""
    endpoints = [
        "https://api.sirketim.com/health",
        "https://auth.sirketim.com/health",
        "https://cdn.sirketim.com/health",
    ]
    for url in endpoints:
        try:
            yanit = requests.get(url, timeout=10)
            if yanit.status_code != 200:
                logging.warning(f"API sagliksiz: {url} -> {yanit.status_code}")
            else:
                logging.info(f"API saglikli: {url}")
        except requests.RequestException as e:
            logging.error(f"API erisim hatasi: {url} -> {e}")

def hizli_ping_kontrol():
    """Hızlı çalışan görev, thread gerektirmez."""
    logging.info("Ping kontrolu yapiliyor...")

# Uzun süren görevi thread'de çalıştır
schedule.every(5).minutes.do(threaded_calistir(uzun_api_kontrolu))

# Hızlı görev normal çalışabilir
schedule.every(1).minutes.do(hizli_ping_kontrol)

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

Görevi Durdurma ve Dinamik Zamanlama

schedule kütüphanesi görevleri dinamik olarak yönetmenize de izin veriyor. Bir görevi belirli koşullarda iptal edebilir ya da yeni görevler ekleyebilirsiniz:

import schedule
import time
import logging

logging.basicConfig(level=logging.INFO)

sayac = 0

def sadece_5_kez_calis():
    global sayac
    sayac += 1
    logging.info(f"Gorev calisti. Sayac: {sayac}")
    
    # 5 kez çalıştıktan sonra kendini iptal et
    if sayac >= 5:
        logging.info("5 kez calisti, gorev iptal ediliyor.")
        return schedule.CancelJob  # Bu dönüş değeri görevi siler

def kosullu_gorev():
    """Sadece mesai saatlerinde çalışsın."""
    from datetime import datetime
    saat = datetime.now().hour
    if 9 <= saat <= 18:
        logging.info("Mesai saatinde calisiyorum.")
    else:
        logging.info("Mesai disi, bu sefer atliyorum.")

def tum_gorevleri_listele():
    logging.info(f"Aktif gorev sayisi: {len(schedule.jobs)}")
    for job in schedule.jobs:
        logging.info(f"  - {job}")

# Sadece 5 kez çalışacak görev
schedule.every(3).seconds.do(sadece_5_kez_calis)

# Mesai saatlerine göre davranan görev
schedule.every(30).minutes.do(kosullu_gorev)

# Görev listesini her saat başı logla
schedule.every().hour.do(tum_gorevleri_listele)

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

systemd ile Servis Olarak Çalıştırma

Script’i sürekli çalışır halde tutmak için systemd servisi oluşturmanız gerekiyor. Cron’dan en büyük fark bu: Script’in ayakta durması lazım. Ama bu aslında bir avantaj, çünkü crash durumunda systemd otomatik yeniden başlatabilir.

# /etc/systemd/system/python-zamanlayici.service dosyasını oluşturun
sudo nano /etc/systemd/system/python-zamanlayici.service

Servis dosyasının içeriği:

[Unit]
Description=Python Schedule Zamanlayici Servisi
After=network.target postgresql.service
Wants=network.target

[Service]
Type=simple
User=deploy
Group=deploy
WorkingDirectory=/opt/myproject
ExecStart=/opt/myproject/venv/bin/python /opt/myproject/zamanlayici.py
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal
SyslogIdentifier=python-zamanlayici

# Ortam değişkenleri
Environment=PYTHONUNBUFFERED=1
EnvironmentFile=/opt/myproject/.env

[Install]
WantedBy=multi-user.target

Sonra servisi etkinleştirip başlatın:

sudo systemctl daemon-reload
sudo systemctl enable python-zamanlayici
sudo systemctl start python-zamanlayici

# Durumu kontrol et
sudo systemctl status python-zamanlayici

# Logları takip et
sudo journalctl -u python-zamanlayici -f

Hata Yönetimi ve Sağlamlık

Üretim ortamında çalışacak bir zamanlayıcıda hata yönetimi kritik önem taşıyor. Görevlerden biri patlasa bile diğerleri çalışmaya devam etmeli:

import schedule
import time
import logging
import traceback
from functools import wraps

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def guvenli_calistir(fonksiyon):
    """Hataları yakalayan decorator."""
    @wraps(fonksiyon)
    def wrapper(*args, **kwargs):
        try:
            return fonksiyon(*args, **kwargs)
        except Exception as e:
            logging.error(
                f"Gorev hatasi [{fonksiyon.__name__}]: {e}n"
                f"{traceback.format_exc()}"
            )
            # İsteğe bağlı: Hata durumunda bildirim gönder
            # mail_gonder(f"Gorev Hatasi: {fonksiyon.__name__}", str(e))
    return wrapper

@guvenli_calistir
def potansiyel_hatali_gorev():
    """Bu görev bazen hata verebilir."""
    import random
    if random.random() < 0.3:
        raise ValueError("Simule edilmis hata!")
    logging.info("Gorev basariyla tamamlandi.")

@guvenli_calistir
def veritabani_temizleme():
    """DB bağlantısı yoksa hata verebilir."""
    import psycopg2
    conn = psycopg2.connect("dbname=uretim user=postgres host=localhost")
    cursor = conn.cursor()
    cursor.execute("DELETE FROM oturumlar WHERE son_erisim < NOW() - INTERVAL '30 days'")
    silinen = cursor.rowcount
    conn.commit()
    conn.close()
    logging.info(f"Eski oturumlar temizlendi: {silinen} kayit")

schedule.every(10).seconds.do(potansiyel_hatali_gorev)
schedule.every().day.at("04:00").do(veritabani_temizleme)

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

schedule vs Cron: Hangisi Ne Zaman?

Her ikisinin de yeri var. Seçim yaparken şunları düşünün:

schedule tercih edin:

• Zamanlama mantığı Python kodunuzla iç içe geçmişse
• Görevler arasında veri paylaşımı gerekiyorsa
• Hata yönetimi ve bildirim sistemi karmaşıksa
• Tek bir uygulama içinde tüm zamanlamayı yönetmek istiyorsanız
• Windows sunucularda çalışıyorsanız (cron yok)
• Dinamik zamanlama ihtiyacınız varsa

Cron tercih edin:

• Basit, bağımsız script’ler çalıştırıyorsanız
• Sistem yeniden başlayınca otomatik devreye girmesi kritikse (systemd ile schedule da bunu yapabilir ama cron daha köklü)
• Birden fazla kullanıcının görevlerini merkezi yönetmek istiyorsanız
• Script’inizin sürekli çalışır halde olmasını istemiyorsanız

schedule’ın kısıtları:

• Script’in sürekli çalışıyor olması gerekiyor
• Sunucu yeniden başlarsa systemd veya benzeri bir mekanizma gerekiyor
• Çok sayıda farklı script için ayrı servisler yönetmek karmaşık hale gelebilir
• Cron kadar köklü değil, edge case’lerde beklenmedik davranışlar görülebilir

Sonuç

schedule kütüphanesi, Python ekosistemi içinde kalan ve insan tarafından okunabilir syntax sunan güzel bir araç. Özellikle Python ile yazılmış uygulamalarda görev zamanlama mantığını kodun içine gömmenin en temiz yolu bu. Cron’un yerini tamamen alması gerekmiyor, aksine her ikisi de araç kutunuzda bulunabilir.

Gerçek üretim senaryolarında en sık kullandığım pattern şu: Basit sistem görevleri için cron, uygulama seviyesindeki karmaşık iş akışları için schedule ile systemd servisi kombinasyonu. Bu ikili yaklaşım hem sadeliği hem de esnekliği beraberinde getiriyor.

Kütüphanenin kaynak kodu oldukça okunabilir ve kısa. Bir ara kaynak koda bakmayı öneririm, nasıl çalıştığını anlamak için birkaç dakika yeterli. Bu tür “kaynak kodunu okuyabileceğiniz” araçlar tercih etmek, özellikle üretim ortamı için çok değerli.

Son olarak: Hangi zamanlama çözümünü kullanırsanız kullanın, loglamayı ihmal etmeyin. Gece 03:00’te sessizce başarısız olan bir görev, sabah onlarca kullanıcı şikayet ederek sizi aradığında farkedilebilir hale geliyor. Erken uyarı her zaman geç özürden iyidir.