Python

How do you safely handle SIGTERM in multi-threaded Python services?

December 3, 2025

download ready
Thank You
Your submission has been received.
We will be in touch and contact you soon!

Use threading.Event() for thread-safe shutdown signal. Main thread catches SIGTERM, sets event. Workers poll event in loops. Daemon threads auto-terminate. 5s join timeout prevents Kubernetes kills.

Code

import signal, threading, time

shutdown = threading.Event()

def handler(sig, frame): shutdown.set()

signal.signal(signal.SIGTERM, handler)

def worker(i):
    while not shutdown.wait(0.1):  # Poll 10x/sec
        time.sleep(0.01)  # Work
    print(f"Worker {i} done")

threads = [threading.Thread(target=worker, args=(i,), daemon=True) 
           for i in range(4)]
for t in threads: t.start()
for t in threads: t.join(5)  # 5s max
      

Test: python app.py & kill %1 → Clean exit <5s

Hire Now!

Need Help with Python Development ?

Work with our skilled python developers to accelerate your project and boost its performance.
**Hire now**Hire Now**Hire Now**Hire now**Hire now

How do you safely handle SIGTERM in multi-threaded Python services?

Use threading.Event() for thread-safe shutdown signal. Main thread catches SIGTERM, sets event. Workers poll event in loops. Daemon threads auto-terminate. 5s join timeout prevents Kubernetes kills.

Code

import signal, threading, time

shutdown = threading.Event()

def handler(sig, frame): shutdown.set()

signal.signal(signal.SIGTERM, handler)

def worker(i):
    while not shutdown.wait(0.1):  # Poll 10x/sec
        time.sleep(0.01)  # Work
    print(f"Worker {i} done")

threads = [threading.Thread(target=worker, args=(i,), daemon=True) 
           for i in range(4)]
for t in threads: t.start()
for t in threads: t.join(5)  # 5s max
      

Test: python app.py & kill %1 → Clean exit <5s