KAHIBARO
Discord Login Register ☰
☰

17.6. Scheduled Tasks

Table of Contents

Why Scheduled Tasks Matter

In many backend systems, some work should run at a specific time or repeatedly, not directly in response to a user request. These are scheduled tasks.

Typical examples:

Doing this inside normal HTTP request handlers is a bad idea. An HTTP request is short lived, but these tasks must run reliably and often when no user is online. Scheduled tasks solve this.

Key idea: A scheduled task is a job that runs automatically at a given time or interval, independent of any single HTTP request.

Scheduled tasks are usually built on top of background jobs and workers, which you met in earlier sections. The scheduler decides when to enqueue a job, and the workers decide how to run it.

Types of Scheduled Tasks

Time based vs Event based

There are two broad ways to think about scheduled work:

This is still a form of scheduling, because you set a task to run in the future, but it is not tied to a fixed time like “every day at midnight.”

Both types can use the same tools. The scheduler library usually supports:

One time vs Recurring

Another useful distinction:

Table comparison:

AspectOne time taskRecurring task
TriggerSpecific date/time onceInterval or cron-like schedule
Example“Run at 2026-12-31 23:59:00”“Run every day at 04:00”
Common useReminders, delayed notificationsMaintenance, reports, data sync

Scheduling with Cron

Before Python tools, it is important to know the classic Unix solution: cron.

What is cron?

On Linux and many Unix systems, cron is a built-in scheduler that runs commands at specified times. A backend application can use cron to:

You define tasks in a crontab (cron table).

Basic crontab line format:

text 
* * * * * command to run
| | | | |
| | | | +----- day of week (0-7, Sunday is 0 or 7)
| | | +------- month (1-12)
| | +--------- day of month (1-31)
| +----------- hour (0-23)
+------------- minute (0-59)

Cron pattern:
minute hour day-of-month month day-of-week

Cron timing examples

ScheduleCron expression
Every minute *
Every day at 03:000 3 *
Every Monday at 09:3030 9 1
Every 15 minutes/15 *
At 00:00 on the 1st of each month0 0 1

You can test your understanding. For example:

Cron then runs the command you specify at those times.

Example: Running a Python script with cron

Imagine you have a Python script:

bash 
# cleanup_sessions.sh
#!/usr/bin/env bash
cd /path/to/app
source venv/bin/activate
python -m app.scripts.cleanup_sessions

You can schedule it:

text 
0 * * * * /path/to/cleanup_sessions.sh >> /var/log/cleanup.log 2>&1

This runs the script at the start of every hour and appends output to a log file.

Pros and cons of cron

Pros:

Cons:

Cron is fine for small projects or for system-level tasks like backups. For application-level tasks that need to integrate with your database and message queues, Python based schedulers are usually better.

Scheduling in Python

Instead of system level cron, you can use Python libraries that run inside your application environment. Common patterns:

Important separation:
Often the scheduler just enqueues jobs, and workers execute jobs.
Do not execute heavy logic directly inside the scheduler loop.

Celery beat: scheduling Celery tasks

If you use Celery for background jobs, Celery beat is the scheduler component.

Architecture:

  1. You define Celery tasks in your application.
  2. You define a beat schedule that says when to run which task.
  3. Celery beat periodically sends tasks to the broker (for example Redis).
  4. Celery workers pick up and execute those tasks.

Basic Celery task

python 
# tasks.py
from celery import Celery
app = Celery("myapp", broker="redis://localhost:6379/0")
@app.task
def cleanup_sessions():
    print("Cleaning up expired sessions...")
    # logic here

Defining periodic tasks

You can define schedule in Python:

python 
from celery.schedules import crontab
app.conf.beat_schedule = {
    "cleanup-sessions-every-hour": {
        "task": "tasks.cleanup_sessions",
        "schedule": crontab(minute=0),  # every hour at minute 0
    },
}

Or using intervals:

python 
app.conf.beat_schedule = {
    "sync-stats-every-5-minutes": {
        "task": "tasks.sync_stats",
        "schedule": 300.0,  # seconds
    },
}

Running beat and workers

Typically, you run:

bash 
celery -A tasks worker --loglevel=info
celery -A tasks beat --loglevel=info

in two different processes or containers.

This keeps scheduling and execution separated.

Adding one time tasks

Celery can also handle delayed one time tasks, for example:

python 
# schedule to run 10 minutes later
from datetime import timedelta
cleanup_sessions.apply_async(countdown=600)

This is more event based, as described earlier.

APScheduler

APScheduler is a pure Python scheduling library. It is often used in:

It supports multiple trigger types:

Basic APScheduler example

python 
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
def send_daily_report():
    print(f"Sending daily report at {datetime.utcnow()}")
scheduler = BackgroundScheduler()
# Run every day at 08:00 UTC
scheduler.add_job(
    send_daily_report,
    trigger="cron",
    hour=8,
    minute=0
)
scheduler.start()
# Keep your main program alive
try:
    while True:
        pass
except KeyboardInterrupt:
    scheduler.shutdown()

This will call send_daily_report every day at 08:00 UTC.

Trigger types in APScheduler

TriggerDescriptionExample
dateRun once at a given datetime“At 2026-01-01 00:00”
intervalRun every N time units“Every 10 minutes”
cronCron style schedule“At 03:00 every day”

Example of an interval job:

python 
# Every 5 minutes
scheduler.add_job(
    sync_external_api,
    trigger="interval",
    minutes=5
)

Example of a one time job:

python 
from datetime import datetime, timedelta
run_at = datetime.utcnow() + timedelta(days=7)
scheduler.add_job(
    remind_subscription,
    trigger="date",
    run_date=run_at,
    args=[user_id]
)

APScheduler with web frameworks

If you run APScheduler inside a web app (for example FastAPI), remember:

Time Zones and Daylight Saving Time

Scheduling across time zones is tricky. If your system has users around the world, you must decide which time zone the scheduler uses.

Common options:

Best practice: Store and process times in UTC internally, convert to local time only at the edges (API / UI).

Example problem

You schedule a daily job at 02:30 local time. When daylight saving time (DST) starts or ends:

Libraries like APScheduler and Python’s zoneinfo (or pytz) can help, but you still need clear rules about what should happen.

Generally:

Idempotency and Safety of Scheduled Jobs

If a scheduled job runs multiple times accidentally, what happens? This is common if:

To keep your system safe, scheduled tasks should usually be idempotent.

Idempotent operation:
An operation is idempotent if running it multiple times has the same effect as running it once.

Non idempotent example

python 
def send_billing_email(user_id):
    # BAD: will send multiple copies if retried
    email_service.send("Monthly bill", to=user_id)

If this job is retried or accidentally scheduled twice, the user gets multiple emails.

More idempotent approach

You can track executions:

python 
def send_billing_email(user_id, billing_period):
    if has_already_sent_email(user_id, billing_period):
        return  # do nothing
    email_service.send("Monthly bill", to=user_id)
    mark_email_sent(user_id, billing_period)

Now, even if the task is executed multiple times, only one email is sent.

Other common strategies:

Monitoring and Managing Scheduled Tasks

Scheduled jobs are long lived. They might run for months or years, so you need to:

Logging and metrics

For each scheduled task, log:

Example:

python 
import logging
from datetime import datetime
logger = logging.getLogger(__name__)
def cleanup_sessions():
    started_at = datetime.utcnow()
    logger.info("cleanup_sessions started", extra={"started_at": started_at.isoformat()})
    try:
        removed = perform_cleanup()
        logger.info(
            "cleanup_sessions finished",
            extra={"removed": removed, "started_at": started_at.isoformat()}
        )
    except Exception:
        logger.exception("cleanup_sessions failed")
        raise

You can also expose metrics like:

These metrics are very helpful when integrated with monitoring tools.

Avoiding overlapping executions

If a task is scheduled every 5 minutes, but sometimes takes 10 minutes to finish, you might have overlapping instances. That can:

Ways to prevent this:

Example using a simple lock idea:

python 
def run_with_lock(lock_key, func, *args, **kwargs):
    if not acquire_lock(lock_key):  # uses Redis or DB
        return  # lock is held, skip this run
    try:
        return func(*args, **kwargs)
    finally:
        release_lock(lock_key)

Then:

python 
def do_daily_aggregation():
    run_with_lock("daily_aggregation_lock", _do_daily_aggregation_impl)

Scheduled Tasks in Practice

To connect everything, here are a few realistic scenarios and how scheduling might look.

Daily cleanup with Celery beat

Goal: Remove users’ expired password reset tokens once per day.

  1. Define the task:
python 
from celery import Celery
from datetime import datetime, timedelta
app = Celery("myapp", broker="redis://localhost:6379/0")
@app.task
def cleanup_password_reset_tokens():
    expiration_time = datetime.utcnow() - timedelta(hours=24)
    # Delete all tokens older than expiration_time
    deleted = (
        db_session.query(PasswordResetToken)
        .filter(PasswordResetToken.created_at < expiration_time)
        .delete()
    )
    db_session.commit()
    print(f"Deleted {deleted} tokens")
  1. Configure beat schedule:
python 
from celery.schedules import crontab
app.conf.beat_schedule = {
    "cleanup-password-reset-tokens-daily": {
        "task": "tasks.cleanup_password_reset_tokens",
        "schedule": crontab(hour=2, minute=0),  # 02:00 UTC every day
    },
}

Now Celery beat schedules the task. Celery workers execute it.

User specific reminders with delayed tasks

Goal: Send a reminder email 3 days before a user’s subscription expires.

In the business logic when the user subscribes:

python 
from datetime import timedelta
def on_user_subscribed(user_id, expires_at):
    reminder_time = expires_at - timedelta(days=3)
    now = datetime.utcnow()
    delay_seconds = max(0, (reminder_time - now).total_seconds())
    send_renewal_reminder.apply_async(
        args=[user_id],
        countdown=delay_seconds
    )

This sets a one time scheduled task for that user.

send_renewal_reminder is a Celery task that checks again that the subscription is still expiring before sending the email, which makes it safer and more idempotent.

Scheduled statistics with APScheduler

Goal: Recalculate daily statistics at midnight.

python 
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
@scheduler.scheduled_job("cron", hour=0, minute=0)
def calculate_daily_stats():
    # aggregate yesterday's data
    pass
if __name__ == "__main__":
    scheduler.start()

You can deploy this as a separate service. It runs indefinitely and only does scheduling and stats.

Summary

With these ideas, you can design backend systems that perform important work reliably and automatically, even when no user is making requests.

Views: 7

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3505 / 24h
Impressum / Imprint
Datenschutz / Data Policy