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

rust 
use serde::Deserialize;
 
#[derive(Debug, Deserialize)]
#[serde(untagged)]

Parsing flexible JSON shapes with serde

deserialization enums json
Intermediate 6 steps
ruby 
require "shellwords"
require "open3"
 
module Backup

Building safe shell commands in Ruby

shell-out subprocess command-injection
Intermediate 7 steps
python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
go 
func (w *Watcher) resetDebounce(d time.Duration) {
	if !w.timer.Stop() {
		select {
		case <-w.timer.C:

Debouncing a stream of events in Go

debounce timers channels
Advanced 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 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