Python Scheduling Techniques: From Simple Loops to APScheduler, Celery, and Airflow

Introduction

In daily development work we often need to run tasks periodically. While Linux crond can be used, Python provides many flexible ways to schedule jobs. This article collects and explains the most common Python scheduling methods.

Table of Contents

Implementing a scheduled task with while True + sleep()

Running scheduled tasks with the Timeloop library

Using threading.Timer for scheduled tasks

Using the built‑in sched module

Using the third‑party schedule module

Implementing scheduled tasks with the APScheduler framework

Using the distributed message system Celery

Using the data‑flow tool Apache Airflow

Implementing a scheduled task with while True + sleep()

The sleep(secs) function from the time module pauses the current thread for secs seconds, after which the thread becomes ready for CPU scheduling. By combining an infinite loop with sleep() we can create a simple periodic task.

Code example:

<code>import datetime
import time

def time_printer():
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func time :', ts)

def loop_monitor():
    while True:
        time_printer()
        time.sleep(5)  # pause 5 seconds

if __name__ == "__main__":
    loop_monitor()
</code>

Main drawbacks:

Can only set an interval, not a specific time (e.g., every day at 08:00).

sleep blocks the thread, so nothing else can run during the pause.

Running scheduled tasks with the Timeloop library

Timeloop is a lightweight library that runs multiple periodic jobs using decorators.

Example code:

<code>import time
from timeloop import Timeloop
from datetime import timedelta

tl = Timeloop()

@tl.job(interval=timedelta(seconds=2))
def sample_job_every_2s():
    print("2s job current time : {}".format(time.ctime()))

@tl.job(interval=timedelta(seconds=5))
def sample_job_every_5s():
    print("5s job current time : {}".format(time.ctime()))

@tl.job(interval=timedelta(seconds=10))
def sample_job_every_10s():
    print("10s job current time : {}".format(time.ctime()))
</code>

Using threading.Timer for scheduled tasks

The Timer class in the threading module is a non‑blocking timer that can start multiple asynchronous tasks.

Signature: Timer(interval, function, args=[], kwargs={})

interval : the delay time

function : the callable to execute

args/kwargs : arguments for the callable

Using the built‑in sched module

The sched.scheduler(timefunc, delayfunc) class provides a generic event scheduler. It requires a time‑returning function (e.g., time.time ) and a delay function (e.g., time.sleep ).

Code example:

<code>import datetime
import time
import sched

def time_printer():
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func time :', ts)

def loop_monitor():
    s = sched.scheduler(time.time, time.sleep)  # create scheduler
    s.enter(5, 1, time_printer, ())
    s.run()

if __name__ == "__main__":
    loop_monitor()
</code>

Key methods of a scheduler object:

enter(delay, priority, action, argument) : schedule an event after delay seconds.

cancel(event) : remove a scheduled event.

run() : execute all pending events.

Using the third‑party schedule module

schedule is a lightweight scheduler that supports seconds, minutes, hours, days, or custom intervals.

Example code:

<code>import schedule
import time

def job():
    print("I'm working...")

schedule.every(10).seconds.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)

while True:
    schedule.run_pending()
    time.sleep(1)
</code>

Features such as passing arguments, using decorators, cancelling jobs, running a job once, and tag‑based job retrieval are also demonstrated in the original article.

Implementing scheduled tasks with the APScheduler framework

APScheduler (Advanced Python Scheduler) is a Quartz‑based framework that supports date‑based, interval‑based, and cron‑style jobs, with persistence options.

Three main features: cron‑like scheduling, interval scheduling, and one‑time execution.

Four core components: Trigger, Jobstore, Executor, Scheduler.

Important concepts:

Job

A job is the smallest executable unit; it stores the callable, arguments, and execution settings such as id , name , trigger , executor , max_instances , next_run_time , misfire_grace_time , coalesce , args , and kwargs .

Trigger

Triggers determine the next run time. APScheduler provides DateTrigger , IntervalTrigger , and CronTrigger with rich parameter sets.

Executor

Executors run jobs; available types include asyncio , gevent , thread pool, process pool, and twisted .

Jobstore

Jobstores persist jobs. Options include in‑memory, MongoDB, Redis, RethinkDB, SQLAlchemy‑backed databases, and Zookeeper.

Scheduler

The scheduler coordinates all components. Common implementations are BlockingScheduler (blocks the main thread) and BackgroundScheduler (runs in the background).

Example code:

<code>from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

sched = BlockingScheduler()
sched.add_job(job, 'interval', seconds=5, id='my_job_id')
sched.start()
</code>

Using the distributed message system Celery

Celery is a flexible distributed task queue that can also schedule periodic jobs. It requires a broker (e.g., RabbitMQ or Redis) and a result backend.

Key components: Celery Beat (scheduler), Producer, Broker, Worker, Result Backend.

Using the data‑flow tool Apache Airflow

Airflow, originally developed at Airbnb, expresses workflows as Directed Acyclic Graphs (DAGs). It supports a rich set of operators (BashOperator, PythonOperator, EmailOperator, HttpOperator, SQL operators, etc.) and can run on various executors such as LocalExecutor , CeleryExecutor , and KubernetesExecutor .

Typical architecture includes a metadata database, scheduler, executor, and workers.

For more details, refer to the original article.