222. Task Scheduling with APScheduler

🔹 1. Installing APScheduler

Copy

pip install apscheduler

✅ Fix: Install APScheduler before using it.

---

🔹 2. Basic APScheduler Job Example

Copy

from apscheduler.schedulers.blocking import BlockingScheduler

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

scheduler = BlockingScheduler()
scheduler.add_job(job, 'interval', seconds=5)  # Runs every 5 seconds
scheduler.start()

✅ Fix: BlockingScheduler() runs indefinitely until manually stopped.

---

🔹 3. Scheduling a Job at a Specific Time

Copy

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime, timedelta

def scheduled_task():
    print("Scheduled task executed!")

scheduler = BlockingScheduler()
run_time = datetime.now() + timedelta(seconds=10)  # Runs 10 seconds from now
scheduler.add_job(scheduled_task, 'date', run_date=run_time)
scheduler.start()

✅ Fix: date trigger runs once at a specific time.

---

🔹 4. Scheduling a Recurring Job (Every Minute)

Copy

from apscheduler.schedulers.background import BackgroundScheduler

def recurring_task():
    print("Recurring task executed!")

scheduler = BackgroundScheduler()
scheduler.add_job(recurring_task, 'interval', minutes=1)  # Runs every 1 minute
scheduler.start()

while True:
    pass  # Keeps the script running

✅ Fix: BackgroundScheduler() runs in non-blocking mode.

---

🔹 5. Using Cron Jobs (Run at a Specific Time Daily)

Copy

from apscheduler.schedulers.blocking import BlockingScheduler

def daily_task():
    print("Daily task executed!")

scheduler = BlockingScheduler()
scheduler.add_job(daily_task, 'cron', hour=10, minute=30)  # Runs daily at 10:30 AM
scheduler.start()

✅ Fix: cron trigger allows precise time-based scheduling.

---

🔹 6. Passing Arguments to Scheduled Functions

Copy

from apscheduler.schedulers.background import BackgroundScheduler

def greet(name):
    print(f"Hello, {name}!")

scheduler = BackgroundScheduler()
scheduler.add_job(greet, 'interval', seconds=10, args=["Alice"])
scheduler.start()

while True:
    pass

✅ Fix: Use args to pass arguments to scheduled functions.

---

🔹 7. Listing All Scheduled Jobs

Copy

from apscheduler.schedulers.background import BackgroundScheduler

def my_task():
    print("Running my task!")

scheduler = BackgroundScheduler()
scheduler.add_job(my_task, 'interval', seconds=10, id='my_task')
scheduler.start()

for job in scheduler.get_jobs():
    print(job)  # Prints all scheduled jobs

✅ Fix: scheduler.get_jobs() retrieves all active jobs.

---

🔹 8. Pausing and Resuming Jobs

Copy

from apscheduler.schedulers.background import BackgroundScheduler

def my_task():
    print("Executing my task!")

scheduler = BackgroundScheduler()
job = scheduler.add_job(my_task, 'interval', seconds=5, id='pause_job')
scheduler.start()

import time
time.sleep(10)
scheduler.pause_job('pause_job')  # Pauses the job
print("Job paused!")

time.sleep(10)
scheduler.resume_job('pause_job')  # Resumes the job
print("Job resumed!")

✅ Fix: pause_job() and resume_job() temporarily disable and enable jobs.

---

🔹 9. Removing a Scheduled Job

Copy

from apscheduler.schedulers.background import BackgroundScheduler

def my_task():
    print("This task will be removed!")

scheduler = BackgroundScheduler()
job = scheduler.add_job(my_task, 'interval', seconds=5, id='remove_job')
scheduler.start()

import time
time.sleep(10)
scheduler.remove_job('remove_job')  # Removes the job
print("Job removed!")

✅ Fix: remove_job('job_id') permanently deletes a job.

---

🔹 10. Stopping the Scheduler Gracefully

Copy

from apscheduler.schedulers.background import BackgroundScheduler
import time

def task():
    print("Executing task...")

scheduler = BackgroundScheduler()
scheduler.add_job(task, 'interval', seconds=5)
scheduler.start()

time.sleep(15)
scheduler.shutdown()  # Stops the scheduler gracefully
print("Scheduler stopped.")

✅ Fix: scheduler.shutdown() stops all tasks safely.

---

🚀 Summary: Why Use APScheduler?

FeatureAPScheduler

One-time execution

✅ date trigger for scheduled one-time jobs

Recurring execution

✅ interval trigger for repeated jobs

Cron-like scheduling

✅ cron trigger for daily/hourly jobs

Background scheduling

✅ BackgroundScheduler() for non-blocking execution

Job control

✅ Pause, resume, or remove jobs dynamically

---

🚀 When to Use APScheduler?

ScenarioUse APScheduler?

Running periodic background tasks

✅ Yes

Replacing cron jobs with Python

✅ Yes

Automating data collection or cleanup

✅ Yes

Triggering scheduled API calls

✅ Yes

Replacing Celery for lightweight scheduling

✅ Yes (but not for distributed tasks)

---