Common Python Scheduling Techniques and Libraries

In daily work we often need to run tasks periodically; this guide collects common Python methods for implementing scheduled jobs, ranging from low‑level loops to full‑featured frameworks.

Using while True + sleep()

The time.sleep(secs) function blocks the current thread for a given number of seconds, allowing a simple infinite loop to act as a scheduler.

<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>

Drawbacks: only interval scheduling, cannot specify exact times, and sleep blocks the thread.

Using the Timeloop library

Timeloop provides a decorator‑based way to run functions at fixed intervals in a separate thread.

<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

The non‑blocking Timer class can schedule a function call after a delay and can be re‑invoked for repeated execution.

<code>import datetime
from threading import Timer

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

def loop_monitor():
    t = Timer(5, time_printer)
    t.start()

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

Note: Timer runs only once; you must call it repeatedly for continuous scheduling.

Using the built‑in sched module

sched.scheduler offers a generic event scheduler that works with any time and delay functions.

<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>

Using the third‑party schedule library

schedule provides a human‑readable syntax for interval, daily, weekly, and specific‑time jobs.

<code>import schedule
import time

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

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

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

Using APScheduler

APScheduler (Advanced Python Scheduler) mirrors Quartz features, supporting date, interval, and cron triggers, persistent job stores, and multiple executors.

Key concepts:

Job – the smallest executable unit.

Trigger – defines when a job runs (date, interval, cron).

Executor – runs jobs (thread pool, process pool, asyncio, etc.).

Jobstore – persists jobs (memory, SQLAlchemy, MongoDB, Redis, etc.).

Event – lifecycle notifications such as job added, executed, missed.

<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 Celery for distributed scheduling

Celery is a robust distributed task queue that can also handle periodic tasks via celery beat . It requires a broker (RabbitMQ, Redis) and a result backend.

Using Apache Airflow

Airflow expresses workflows as Directed Acyclic Graphs (DAGs) of tasks, supporting rich operators (BashOperator, PythonOperator, EmailOperator, etc.) and a pluggable executor architecture (Sequential, Local, Celery, Dask, Kubernetes).

Key components:

Metadata database – stores DAG and task state.

Scheduler – decides which tasks to run based on DAG definitions.

Executor – launches task instances (e.g., CeleryExecutor, KubernetesExecutor).

Workers – actual processes that execute tasks.

Airflow enables complex dependencies, branching, and monitoring via a web UI.

Original source: https://www.biaodianfu.com/python-schedule.html