python 44 lines · 6 steps

A self-rescheduling periodic task in Python

A thread-safe helper that runs a function on a repeating interval using chained one-shot timers.

Explained by highlit
Intermediate threading scheduling concurrency locks error-handling
1import threading
2import logging
3 
4logger = logging.getLogger(__name__)
5 
6 
7class PeriodicTask:
8 def __init__(self, interval, func, *args, **kwargs):
9 self.interval = interval
10 self.func = func
11 self.args = args
12 self.kwargs = kwargs
13 self._timer = None
14 self._lock = threading.Lock()
15 self._running = False
16 
17 def start(self):
18 with self._lock:
19 if self._running:
20 return
21 self._running = True
22 self._schedule()
23 
24 def _schedule(self):
25 self._timer = threading.Timer(self.interval, self._run)
26 self._timer.daemon = True
27 self._timer.start()
28 
29 def _run(self):
30 try:
31 self.func(*self.args, **self.kwargs)
32 except Exception:
33 logger.exception("periodic task %r failed", self.func.__name__)
34 finally:
35 with self._lock:
36 if self._running:
37 self._schedule()
38 
39 def stop(self):
40 with self._lock:
41 self._running = False
42 if self._timer is not None:
43 self._timer.cancel()
44 self._timer = None
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Chaining one-shot timers gives a repeating schedule that measures the interval between the end of one run and the start of the next.
  2. 2A lock guarding the running flag prevents start/stop races from leaving stray timers scheduled.
  3. 3Catching exceptions inside the task keeps one failure from silently killing the whole repeating loop.

Related explainers

python 
from collections.abc import MutableMapping
 
 
class CaseInsensitiveDict(MutableMapping):

Building a case-insensitive dict in Python

data structures abstract base classes dunder methods
Intermediate 8 steps
java 
public class TimedFetchService {
 
    private final ExecutorService executor = Executors.newFixedThreadPool(8);
    private final HttpClient httpClient = HttpClient.newHttpClient();

Enforcing HTTP timeouts with a Future

concurrency timeouts thread-pool
Intermediate 8 steps
python 
from fastapi import FastAPI, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

Redacting sensitive fields in FastAPI errors

validation error-handling security
Intermediate 7 steps
python 
from datetime import timedelta
 
from flask import Blueprint, current_app, make_response, redirect, request, url_for
from itsdangerous import URLSafeTimedSerializer, BadSignature, SignatureExpired

How signed remember-me cookies work in Flask

authentication signed-cookies session-management
Intermediate 8 steps
javascript 
import { unstable_cache, revalidateTag } from 'next/cache'
import { db } from '@/lib/db'
 
export const getDashboardStats = unstable_cache(

Caching dashboard stats in Next.js

caching cache-invalidation tag-based-revalidation
Intermediate 8 steps
python 
import click
from flask.cli import AppGroup
 
from . import db

Custom user CLI commands in Flask

cli click command-group
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

A self-rescheduling periodic task in Python — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code