Zamanlama ve Cron Tabanlı Aracılar
APScheduler, cron görevleri ve zaman tetiklemeli otonom aracı yürütmesi.
Zamanlama ve Cron Tabanlı Aracılar, CoddyKit'te ücretsiz bir AI Agents dersidir. Bu, 4 dersinin 3. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, AI Agents öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. AI Agents kursu toplamda 4 dersten oluşur.
Ajanlar Neden Zamanlanır?
Bazı ajan görevlerinin sürekli veya isteğe bağlı olarak çalışması gerekmez — bunlar bir zamanlamaya göre çalışır. Günlük raporlar, haftalık özetler, saatlik veri kontrolleri ve dönemsel temizlikler, zamanlanan ajanlar için uygun görevlerdir.
APScheduler Temelleri
APScheduler (Gelişmiş Python Zamanlayıcısı), işleri Python işleminizin içinde çalıştırır. Üç tetikleyici türünü destekler: date (bir kez), interval (tekrarlanan) ve cron (takvim tabanlı).
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime
# BlockingScheduler: takes over the main thread
# BackgroundScheduler: runs in background thread
scheduler = BackgroundScheduler()
def my_agent_job():
print(f'Agent running at {datetime.now()}')
# Add a simple interval job
scheduler.add_job(my_agent_job, 'interval', minutes=5)
scheduler.start()
print('Scheduler started in background')
# Your app continues running here
import time
time.sleep(15)
scheduler.shutdown()
print('Scheduler stopped')Cron Tetikleyicisi Söz Dizimi
Cron tetikleyicisi, alışıldık cron alanlarını kullanır: year, month, day, week, day_of_week, hour, minute, second. Sayıları, aralıkları, listeleri ve joker karakterleri kullanabilirsiniz.
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
def morning_briefing():
print('Good morning! Running daily briefing agent')
def weekly_report():
print('Running weekly summary')
def every_business_hour():
print('Hourly check during business hours')
# Every day at 9:00 AM
scheduler.add_job(morning_briefing, 'cron', hour=9, minute=0)
# Every Monday at 8:30 AM
scheduler.add_job(weekly_report, 'cron', day_of_week='mon', hour=8, minute=30)
# Every hour from 9am to 5pm, weekdays only
scheduler.add_job(every_business_hour, 'cron',
day_of_week='mon-fri',
hour='9-17',
minute=0
)
scheduler.start()
print('Jobs scheduled:', len(scheduler.get_jobs()))Cron İfadesi Söz Dizimi
Standart cron ifadeleri 5 alan kullanır: minute hour day month weekday. APScheduler, bunları CronTrigger.from_crontab() ile tek bir dize olarak da kabul eder.
0 9 * * *— her gün saat 09.000 9 * * 1— her pazartesi saat 09.00*/15 * * * *— her 15 dakikada bir
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
scheduler = BackgroundScheduler()
def agent_task():
print('Running scheduled agent')
# From crontab string: every day at 9am
trigger = CronTrigger.from_crontab('0 9 * * *')
scheduler.add_job(agent_task, trigger)
# Equivalent explicit form
scheduler.add_job(
agent_task,
'cron',
minute=0,
hour=9
)
# Every 15 minutes
scheduler.add_job(agent_task, CronTrigger.from_crontab('*/15 * * * *'))
print('Scheduled jobs:')
for job in scheduler.get_jobs():
print(f' {job.id}: next run {job.next_run_time}')Aralık Tetikleyicisi
Aralık tetikleyicisi, bir işi her N zaman biriminde bir çalıştırır. Düzenli sorgulama, canlılık sinyalleri veya sabit bir sıklıkta tekrarlanması gereken tüm görevler için kullanın.
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime, timedelta
scheduler = BackgroundScheduler()
def check_for_updates():
print(f'Checking for updates at {datetime.now()}')
# Agent logic: poll API, check for new items
# Every 30 minutes
scheduler.add_job(check_for_updates, 'interval', minutes=30)
# Every 2 hours, starting 10 minutes from now
start_time = datetime.now() + timedelta(minutes=10)
scheduler.add_job(
check_for_updates,
'interval',
hours=2,
start_date=start_time
)
# Run once in the future (date trigger)
from apscheduler.triggers.date import DateTrigger
run_at = datetime.now() + timedelta(minutes=5)
scheduler.add_job(check_for_updates, DateTrigger(run_date=run_at))
scheduler.start()
print('All jobs scheduled')İş Parametreleri ve ID'ler
Daha sonra işlere başvurabilmek, onları duraklatabilmek veya kaldırabilmek için işlere ID'ler atayın. Bağımsız değişkenleri iş işlevine args veya kwargs üzerinden geçirin.
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
def fetch_report(report_type, user_id):
print(f'Fetching {report_type} report for user {user_id}')
# Named job with arguments
scheduler.add_job(
fetch_report,
'cron',
hour=9,
minute=0,
id='daily_report_user_42',
kwargs={'report_type': 'daily', 'user_id': 42},
replace_existing=True # Update if job already exists
)
scheduler.start()
# Pause a specific job
scheduler.pause_job('daily_report_user_42')
print('Job paused')
# Resume it
scheduler.resume_job('daily_report_user_42')
print('Job resumed')
# Remove it
scheduler.remove_job('daily_report_user_42')
print('Job removed')Yeniden Başlatmalar Arasında İşleri Kalıcılaştırma
Varsayılan olarak APScheduler, işleri bellekte saklar ve yeniden başlatıldığında bu işler kaybolur. İşlerin yeniden başlatmalardan sonra da korunması için onları bir veritabanında kalıcılaştıran SQLAlchemy iş deposunu kullanın.
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor
jobstores = {
'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
'default': ThreadPoolExecutor(20)
}
scheduler = BackgroundScheduler(
jobstores=jobstores,
executors=executors
)
def persistent_agent():
print('Running persisted scheduled agent')
# This job survives restarts
scheduler.add_job(
persistent_agent,
'cron',
hour=8,
minute=0,
id='morning_agent',
replace_existing=True
)
scheduler.start()
print('Scheduler started with SQLite persistence')İş İstisnalarını Ele Alma
Gelecekteki çalıştırmaların tek bir başarısız çalıştırma nedeniyle sessizce durmasını önlemek için zamanlanmış iş işlevlerini try/except ile sarın. Başarısızlıkları günlüğe kaydedin ve isteğe bağlı olarak uyarılar gönderin.
import logging
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('scheduler')
scheduler = BackgroundScheduler()
def job_listener(event):
if event.exception:
logger.error(f'Job {event.job_id} failed: {event.exception}')
# Optionally send alert: email, Slack, PagerDuty
else:
logger.info(f'Job {event.job_id} completed successfully')
scheduler.add_listener(job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
def my_agent_job():
# Errors here are caught by the listener
raise ValueError('Something went wrong in the agent')
scheduler.add_job(my_agent_job, 'interval', seconds=10, id='test_job')
scheduler.start()İş Çakışmalarını Önleme
Bir iş kendi aralığından daha uzun sürerse sonraki çalıştırma, önceki çalıştırma tamamlanmadan başlayabilir. max_instances=1 (varsayılan değerini) ayarlayın veya kaçırılan çalıştırmaları atlamak için birleştirme kullanın.
from apscheduler.schedulers.background import BackgroundScheduler
import time
scheduler = BackgroundScheduler()
def slow_agent():
print('Agent started')
time.sleep(45) # Takes 45 seconds
print('Agent finished')
# max_instances=1: only one run at a time (default)
# coalesce=True: if multiple runs were missed, fire only once when caught up
scheduler.add_job(
slow_agent,
'interval',
minutes=1,
max_instances=1,
coalesce=True,
id='slow_agent'
)
scheduler.start()
print('Slow agent scheduled (max 1 concurrent run)')Zamanlamayı FastAPI ile Tümleştirme
Zamanlayıcıyı sunucunun yaşam döngüsüyle düzgün biçimde başlatıp durdurmak için FastAPI içinde yaşam döngüsü bağlamıyla birlikte APScheduler'ın BackgroundScheduler'ını kullanın.
from fastapi import FastAPI
from apscheduler.schedulers.background import BackgroundScheduler
from contextlib import asynccontextmanager
scheduler = BackgroundScheduler()
def morning_agent_job():
print('Morning agent running')
@asynccontextmanager
async def lifespan(app: FastAPI):
# Start scheduler when app starts
scheduler.add_job(morning_agent_job, 'cron', hour=9, minute=0)
scheduler.start()
print('Scheduler started')
yield # App runs here
# Stop scheduler when app shuts down
scheduler.shutdown()
print('Scheduler stopped')
app = FastAPI(lifespan=lifespan)
@app.get('/jobs')
def list_jobs():
return [
{'id': job.id, 'next_run': str(job.next_run_time)}
for job in scheduler.get_jobs()
]Saat Dilimine Duyarlı Zamanlama
Zamanlanan ajanlar için saat dilimlerini her zaman belirtin. Saat dilimi bağlamı olmayan cron işleri, yaz saati uygulamasındaki değişikliklerden sonra yanlış zamanlarda çalışabilir.
from apscheduler.schedulers.background import BackgroundScheduler
import pytz
# Scheduler with timezone
scheduler = BackgroundScheduler(timezone='America/New_York')
def ny_morning_agent():
print('Running at 9am New York time')
def london_eod_agent():
print('Running at 5pm London time')
# 9am New York (handles EST/EDT automatically)
scheduler.add_job(
ny_morning_agent,
'cron',
hour=9,
minute=0,
timezone=pytz.timezone('America/New_York')
)
# 5pm London time
scheduler.add_job(
london_eod_agent,
'cron',
hour=17,
minute=0,
timezone=pytz.timezone('Europe/London')
)
scheduler.start()
print('Timezone-aware scheduler running')Bilgi Kontrolü: Zamanlama
APScheduler'ı ve cron tabanlı ajanları anlayıp anlamadığınızı sınayın.
Zamanlama için En İyi Uygulamalar
Güvenilir zamanlanan ajanlar için temel kurallar:
- Saat dilimine duyarlı zamanlamayı her zaman kullanın
- Yeniden başlatmalardan sonra korunmaları için işleri bir veritabanında kalıcılaştırın
- Uzun süren işler için
max_instances=1ayarlayın - Sessiz başarısızlıkları yakalamak için hata dinleyicileri ekleyin
- Her zamanlanan çalıştırmayı başlangıç zamanı, bitiş zamanı ve sonucuyla birlikte günlüğe kaydedin
Sıkça Sorulan Sorular
“Zamanlama ve Cron Tabanlı Aracılar” dersi ücretsiz mi?
Evet — “Zamanlama ve Cron Tabanlı Aracılar” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve AI Agents kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. AI Agents kursu toplamda 4 dersten oluşur.
“Zamanlama ve Cron Tabanlı Aracılar” dersinde ne öğreneceğim?
APScheduler, cron görevleri ve zaman tetiklemeli otonom aracı yürütmesi. AI Agents ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.
AI Agents öğrenmeye başlamak için deneyim gerekli mi?
Önceden deneyim gerekmez. CoddyKit'te AI Agents, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 3. dersidir.
“Zamanlama ve Cron Tabanlı Aracılar” dersi ne kadar sürer?
Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.
Bu AI Agents dersinde kod yazıp çalıştırabilir miyim?
Evet. Her AI Agents dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.
Bu kursun tüm dersleri
- Tetikleyici-Eylem Aracı Kalıpları
- Aracıları Web Kancalarına Bağlama
- Zamanlama ve Cron Tabanlı Aracılar
- Birden Fazla Uygulama İçin Otomasyon Hattı Oluşturma