Safe Threading with SQLAlchemy in FastAPI

When building FastAPI applications that use SQLAlchemy for data persistence, you often need to perform background tasks—such as updating records or processing data—without blocking the main request. A common approach is to offload work to background threads. However, threading in Python comes with its own challenges, especially regarding SQLAlchemy sessions.

In this article, we are going to understand the limitations of using threads with SQLAlchemy sessions, and then we’ll discuss how to safely address these challenges.

What Are Threads?

Threads are a way to perform multiple operations concurrently within a single process. They allow your application to handle tasks in parallel—such as processing I/O-bound operations—without waiting for one task to finish before starting another.

Usage in FastAPI:

In a FastAPI application, you might use threads to run background tasks so that the main API request isn’t delayed. For example, sending an email, updating a record, or doing non-critical processing.

Limitations of Threads

Not Truly Parallel (CPython):
Due to the Global Interpreter Lock (GIL), threads in CPython are most effective for I/O-bound tasks. CPU-bound tasks might not gain much speed-up.

Shared State and Concurrency:
Threads share the same memory space, which can lead to complications when mutable shared state is not managed carefully.

Thread Safety Issues:
Not every component is thread-safe. In particular, SQLAlchemy sessions are not thread-safe—using the same session in multiple threads can lead to unpredictable behavior and errors.

The Solution: Isolate SQLAlchemy Sessions in Threads

Instead of passing the existing session to a background thread, you should create a new session within that thread. This ensures that each thread has its own isolated session and avoids the discussed pitfalls.

Implementation Steps

Define a Session Factory:
In your application, use SQLAlchemy’s sessionmaker to create a session factory. For example:

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Use the Factory in Background Tasks:

When you need to run a background task (e.g., updating a record), create a new session inside that task:

# background_tasks.py

from database import SessionLocal
from config.logger_config import logger

def update_record_background(param):
    db = SessionLocal()  # Create a new session for this thread
    try:
        # Perform your database operations here
        # e.g., update a record or commit a transaction
        db.commit()
    except Exception as e:
        db.rollback()
        logger.error(f"Error in background update: {e}")
    finally:
        db.close()

Launch the Background Task:

In your FastAPI endpoint or wherever the background task is needed, start a new thread without passing the main session:

import threading
from background_tasks import update_record_background

# When triggering the background task:
threading.Thread(target=update_record_background, args=(param_value,)).start()

Benefits of This Approach

Thread Safety:
Each thread operates on its own session, eliminating race conditions and ensuring data consistency.

Isolation:
Background tasks do not interfere with the main request’s session, preventing unexpected state changes.

Scalability:
This approach makes your application easier to scale, as each task uses an independent session.

We’ve identified the core issues when sharing SQLAlchemy sessions between threads: thread safety concerns and unexpected state changes. The solution is to create a new session for each background task, ensuring isolation and stability in your FastAPI applications.

Implementing this pattern not only resolves errors like IllegalStateChangeError but also leads to cleaner, more maintainable code.
This strategy is applicable to any FastAPI application using SQLAlchemy and can be adapted for more complex asynchronous tasks using task queues if needed.