I Eliminated SQLite Race Conditions in a Multi-Threaded Python App 🚀

Source: Dev.to

Introduction

Random crashes, database corruption, and “database is locked” errors plagued my app Sortify when multiple threads accessed SQLite simultaneously. This post explains how I eliminated those race conditions and made the database production‑ready.

The Problem

• A single SQLite connection was shared across several concurrent components:

  • Auto‑sort watcher
  • Manual file operations
  • Scheduler tasks
  • Background processing threads

• Symptoms:

  • Random crashes
  • database is locked errors
  • Inconsistent history data
  • Risk of database corruption
  • General app instability under load

The culprit was this line:

sqlite3.connect(db_path, check_same_thread=False)

check_same_thread=False disables SQLite’s built‑in thread safety without actually making the connection thread‑safe. SQLite allows multiple connections, but each connection must stay in the thread that created it. Sharing cursors, connections, or transactions across threads leads to race conditions.

Solution Architecture

I introduced a thread‑local database manager that provides each thread with its own safe connection, automatic retry logic, and a centralized access layer.

Thread‑Local Connections

# core/database_manager.py
import threading
import sqlite3
import time
from typing import Callable, Any, List

class DatabaseManager:
    def __init__(self, db_path: str, timeout: float = 10.0):
        self.db_path = db_path
        self.timeout = timeout
        self._local = threading.local()

    def _get_connection(self) -> sqlite3.Connection:
        """Create or retrieve the connection bound to the current thread."""
        if not hasattr(self._local, "conn"):
            self._local.conn = sqlite3.connect(
                self.db_path,
                timeout=self.timeout,
                check_same_thread=True  # ✅ safe
            )
        return self._local.conn

    def execute(self, query: str, params: tuple = ()) -> sqlite3.Cursor:
        """Execute a single statement with automatic retry on lock."""
        return self._retry(lambda: self._get_connection().execute(query, params))

    def execute_transaction(self, operations: List[Callable[[sqlite3.Connection], Any]]) -> None:
        """Run a series of operations atomically."""
        conn = self._get_connection()
        try:
            conn.execute("BEGIN")
            for op in operations:
                op(conn)
            conn.commit()
        except Exception:
            conn.rollback()
            raise

    def close_all_connections(self) -> None:
        """Close the connection for the current thread."""
        if hasattr(self._local, "conn"):
            self._local.conn.close()
            del self._local.conn

    # ---- internal helpers -------------------------------------------------
    def _retry(self, func: Callable[[], Any], retries: int = 5, backoff: float = 0.1) -> Any:
        """Retry a DB operation when the database is locked."""
        for attempt in range(retries):
            try:
                return func()
            except sqlite3.OperationalError as e:
                if "database is locked" not in str(e):
                    raise
                time.sleep(backoff * (2 ** attempt))
        raise sqlite3.OperationalError("Maximum retries exceeded: database is locked")

How It Works

• Thread‑local storage (threading.local()) ensures each thread gets its own sqlite3.Connection.
• Connections are created on demand and reused within the same thread.
• check_same_thread=True enforces SQLite’s safety guarantees.
• The execute method automatically retries operations that fail with database is locked, using exponential back‑off.
• execute_transaction guarantees atomic writes even under heavy load.
• close_all_connections prevents leaked file handles.

Refactoring Existing Code

# Example usage in other modules
from .database_manager import DatabaseManager

db_manager = DatabaseManager("/path/to/sortify.db")

# Simple query
cursor = db_manager.execute("SELECT * FROM history WHERE id = ?", (record_id,))
rows = cursor.fetchall()

# Transactional update
def delete_history(conn):
    conn.execute("DELETE FROM history WHERE user_id = ?", (user_id,))

db_manager.execute_transaction([delete_history])

All database interactions now pass through a single, safe gateway, eliminating hidden race conditions.

Stress Testing

To verify thread safety, I wrote a stress‑test suite:

# tests/test_database_threading.py
import threading
from core.database_manager import DatabaseManager

def worker(db_path, ops):
    mgr = DatabaseManager(db_path)
    for i in range(ops):
        mgr.execute("INSERT INTO test (value) VALUES (?)", (i,))

def test_threaded_writes():
    db_path = "test.db"
    threads = []
    for _ in range(5):
        t = threading.Thread(target=worker, args=(db_path, 50))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()

    # Verify results
    mgr = DatabaseManager(db_path)
    count = mgr.execute("SELECT COUNT(*) FROM test").fetchone()[0]
    assert count == 250

Results

The test produced zero failures, zero locks, and no corruption.

File Changes Overview

Key Takeaways

• SQLite is thread‑friendly, not thread‑safe.
• check_same_thread=False is a common trap; avoid it unless you implement your own locking.
• The correct model is one connection per thread.
• Centralizing DB access through a manager prevents future bugs and simplifies maintenance.
• Stress tests that simulate real concurrency are essential; unit tests alone won’t expose race conditions.

Repository

The full implementation is available on GitHub:

https://github.com/Mrtracker-new/Sortify

Conclusion

If your Python application uses SQLite and runs background threads, the pattern above will protect you from random crashes, locked‑database errors, and data corruption. Implement a thread‑local DatabaseManager, add retry logic, and centralize all DB interactions for a stable, production‑ready solution. Happy coding! 🚀