Signal Handling in Python Safely

When you're building a Python application that needs to run for a long time—like a web server, a background worker, or a data pipeline—you'll eventually need to handle signals. Signals are those little interrupts the operating system sends to your process, like when someone presses Ctrl+C (SIGINT) or when the system wants to shut you down gracefully.

I remember when I first started working with signals in Python, I thought it was just a matter of catching them and moving on. But as PythonSkillset's documentation shows, there's a lot more nuance to doing it safely. Let's break down the common gotchas and how to handle signals without breaking your code.

Why Signal Handling Can Be Tricky

Python's signal handling runs on the main thread and interrupts whatever your code is doing at that exact moment. This is fundamentally different from threading or async code where you control when things happen. The signal handler gets called in the middle of whatever your main thread is executing.

Here's a classic example that seems straightforward but can cause problems:

import signal
import time

should_stop = False

def handle_signal(signum, frame):
    global should_stop
    should_stop = True

signal.signal(signal.SIGINT, handle_signal)

while not should_stop:
    # Your main loop
    print("Working...")
    time.sleep(2)

This looks fine, right? But imagine you're not just printing—you're writing to a file, updating a database, or manipulating a complex data structure. When the signal arrives, your code could be in the middle of any operation. Setting should_stop = True happens instantly, but your main loop might not check that variable until it's too late.

The Real Danger: Race Conditions

The biggest problem with naive signal handling is that Python's signal handlers run in the main thread, but they don't hold the GIL in a special way. If your main loop is in the middle of a non-atomic operation—like appending to a list, writing to a file, or modifying a shared resource—the signal can interrupt it.

Let's look at a more realistic scenario from PythonSkillset's production code:

import signal
import json
import time

data_buffer = []

def handle_signal(signum, frame):
    print("Saving buffer...")
    with open("backup.json", "w") as f:
        json.dump(data_buffer, f)
    exit(0)

signal.signal(signal.SIGINT, handle_signal)

while True:
    # Simulate collecting data
    new_item = {"timestamp": time.time(), "value": 42}
    data_buffer.append(new_item)
    time.sleep(0.1)

If you press Ctrl+C while data_buffer is being modified, the signal handler tries to save an inconsistent state. The list might be in the middle of resizing, or the newly appended item might not be fully written yet. You'd end up with a corrupted backup file.

Enter: Signals in a Queue

PythonSkillset's recommended approach is to decouple signal handling from your main logic. Instead of setting a flag directly, push the signal onto a queue and let your main loop handle it when it's safe.

Here's how that looks:

import signal
import queue
import time
import json

signal_queue = queue.Queue()

def handle_signal(signum, frame):
    signal_queue.put(signum)

signal.signal(signal.SIGINT, handle_signal)
signal.signal(signal.SIGTERM, handle_signal)

data_buffer = []

def process_signals():
    while not signal_queue.empty():
        sig = signal_queue.get()
        if sig in (signal.SIGINT, signal.SIGTERM):
            print(f"Received signal {sig}, shutting down gracefully...")
            with open("backup.json", "w") as f:
                json.dump(data_buffer, f)
            exit(0)

while True:
    # Do your actual work
    new_item = {"timestamp": time.time(), "value": 42}
    data_buffer.append(new_item)

    # Check for signals at a safe point
    process_signals()

    time.sleep(0.1)

This way, the signal handler only puts the signal number into a queue—which is a fast, atomic operation. Your main loop checks the queue at well-defined points where it's safe to shut down. The data buffer won't be corrupted because you only save it between appends.

But Wait—What About Threads?

If you're using threads, things get even more interesting. Python's signal handlers always run in the main thread, but your worker threads might be doing important work too. The same queue pattern works, but you need to coordinate between threads.

PythonSkillset's threading guide suggests using a threading event:

import signal
import threading
import time

shutdown_event = threading.Event()

def handle_signal(signum, frame):
    shutdown_event.set()

signal.signal(signal.SIGINT, handle_signal)
signal.signal(signal.SIGTERM, handle_signal)

def worker():
    while not shutdown_event.is_set():
        print(f"{threading.current_thread().name} working...")
        time.sleep(1)

threads = []
for i in range(4):
    t = threading.Thread(target=worker, name=f"Worker-{i}")
    t.start()
    threads.append(t)

# Main thread waits for shutdown
shutdown_event.wait()
print("Shutting down all threads...")
for t in threads:
    t.join(timeout=5)

The beauty here is that shutdown_event.set() is atomic and thread-safe. All workers check the event at the top of their loop, so they'll stop cleanly. No corrupted state, no half-written files.

Handling Signals in Async Code

Async Python has its own quirks with signals. If you're using asyncio, you can't just use signal.signal() because it might interfere with the event loop. Instead, you should use loop.add_signal_handler().

Here's the PythonSkillset-approved way:

import asyncio
import signal

async def main():
    shutdown_event = asyncio.Event()

    def handle_signal():
        print("Shutdown requested")
        shutdown_event.set()

    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, handle_signal)
    loop.add_signal_handler(signal.SIGTERM, handle_signal)

    async def worker():
        while not shutdown_event.is_set():
            await asyncio.sleep(1)
            print("Working...")

    tasks = [asyncio.create_task(worker()) for _ in range(3)]

    await shutdown_event.wait()
    print("Shutting down...")
    for task in tasks:
        task.cancel()

asyncio.run(main())

This keeps signals out of your async functions entirely. The event loop handles the signal in its own context, setting the event which your coroutines can check at await points.

Practical Tips from PythonSkillset

After dealing with signal handling in many production systems, here are the patterns that work:

1. Never do I/O in your signal handler—Signal handlers run with minimal context. Writing to files, making network calls, or even printing can cause deadlocks or crashes.

2. Use signal.set_wakeup_fd() for complex setups—If you're building something like a server with multiple event sources, you can write to a file descriptor from the signal handler and monitor that fd in your main loop.

3. Test your shutdown paths—Don't assume your graceful shutdown works. Trigger signals in your tests and verify that files are saved, connections are closed, and state is consistent.

4. Consider SIGHUP for reloads—Unix systems use SIGHUP to signal processes to reload configuration. You can use the same queue pattern to trigger a configuration reload without restarting the whole process.

A Complete Example

Let me put it all together with a practical example from PythonSkillset's monitoring tool:

import signal
import queue
import json
import time
import sys
from datetime import datetime

class GracefulShutdown:
    def __init__(self):
        self.signal_queue = queue.Queue()
        self._register_signals()

    def _register_signals(self):
        signal.signal(signal.SIGINT, self._handler)
        signal.signal(signal.SIGTERM, self._handler)
        signal.signal(signal.SIGHUP, self._handler)

    def _handler(self, signum, frame):
        self.signal_queue.put(signum)

    def check(self, data_buffer):
        try:
            signum = self.signal_queue.get_nowait()
            if signum == signal.SIGHUP:
                print("Reloading configuration...")
                # Reload logic here
            else:
                print(f"Received {signum}, saving state...")
                self._save_state(data_buffer)
                sys.exit(0)
        except queue.Empty:
            pass

    def _save_state(self, data):
        filename = f"state_{datetime.now().isoformat()}.json"
        with open(filename, "w") as f:
            json.dump(data, f)

shutdown = GracefulShutdown()
data = []

for i in range(1000):
    data.append({"index": i, "time": time.time()})
    shutdown.check(data)
    time.sleep(0.5)

print("Completed normally")

This pattern has saved me from corrupted data more times than I can count. The key takeaway is simple: signals should be the messenger, not the decision-maker. Let your application logic decide when and how to act on them.

If you're building anything that needs to run reliably, take the time to handle signals properly. Your future self—and your users—will thank you when the system shuts down gracefully instead of leaving a mess behind.