I'm building a multi-tenant FastAPI application that uses PostgreSQL schemas to separate tenant data. I have a middleware that extracts an X-Tenant-ID header, looks up the tenant's schema, and then switches the current schema for the database session accordingly. For a single request (via Postman) the middleware works fine; however, when sending multiple requests concurrently, I sometimes get errors such as:
It appears that the DB connection is closing prematurely or reverting to the public schema too soon, so tenant-specific tables are not found.
Below are the relevant code snippets:
SchemaSwitchMiddleware)from typing import Optional, Callable
from fastapi import Request, Response
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from app.db.session import SessionLocal, switch_schema
from app.repositories.tenant_repository import TenantRepository
from app.core.logger import logger
from contextvars import ContextVar
current_schema: ContextVar[str] = ContextVar("current_schema", default="public")
class SchemaSwitchMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next: Callable) -> Response:
"""
Middleware to dynamically switch the schema based on the `X-Tenant-ID` header.
If no header is present, defaults to `public` schema.
"""
db = SessionLocal() # Create a session here
try:
tenant_id: Optional[str] = request.headers.get("X-Tenant-ID")
if tenant_id:
try:
tenant_repo = TenantRepository(db)
tenant = tenant_repo.get_tenant_by_id(tenant_id)
if tenant:
schema_name = tenant.schema_name
else:
logger.warning("Invalid Tenant ID received in request headers")
return JSONResponse(
{"detail": "Invalid access"},
status_code=400
)
except Exception as e:
logger.error(f"Error fetching tenant: {e}. Defaulting to public schema.")
db.rollback()
schema_name = "public"
else:
schema_name = "public"
current_schema.set(schema_name)
switch_schema(db, schema_name)
request.state.db = db # Store the session in request state
response = await call_next(request)
return response
except Exception as e:
logger.error(f"SchemaSwitchMiddleware error: {str(e)}")
db.rollback()
return JSONResponse({"detail": "Internal Server Error"}, status_code=500)
finally:
switch_schema(db, "public") # Always revert to public
db.close()
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, declarative_base, Session
from app.core.logger import logger
from app.core.config import settings
# Base for models
Base = declarative_base()
DATABASE_URL = settings.DATABASE_URL
# SQLAlchemy engine
engine = create_engine(
DATABASE_URL,
pool_pre_ping=True,
pool_size=20,
max_overflow=30,
)
# Session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def switch_schema(db: Session, schema_name: str):
"""Helper function to switch the search_path to the desired schema."""
db.execute(text(f"SET search_path TO {schema_name}"))
db.commit()
# logger.debug(f"Switched schema to: {schema_name}")
Public Schema: Contains tables like users, roles, tenants, and user_lookup.
Tenant Schema: Contains tables like users, roles, buildings, and floors.
When I test with a single request, everything works fine. However, with concurrent requests, the switching sometimes reverts to the public schema too early, resulting in errors because tenant-specific tables are missing.
any help on this is much apricated. Thankyou
I also implemented a multi-tenant FastAPI application using PostgreSQL via schemas.
In my case, I avoided using middleware because the database session (obtained from SessionLocal) and its state need to be isolated per request.
When using middleware, the connection (and its state) from the connection pool may be reused across requests. Even though ContextVar is designed to be isolated in asynchronous contexts, the actual database connection can still be shared, leading to race conditions. For example, if one request changes the schema and the connection is then reused for another request, that new request might unexpectedly start with the wrong schema (like reverting to "public").
Instead, I handle the tenant schema switching in a dependency (using Depends). This way, each request gets its own session, and we can safely set the schema for that specific request without affecting others.
Below is an example implementation using a synchronous SQLAlchemy Session
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, declarative_base, Session
from app.core.logger import logger
from app.core.config import settings
from typing import Annotated
from fastapi import Header
# Base for models
Base = declarative_base()
DATABASE_URL = settings.DATABASE_URL
# SQLAlchemy engine
engine = create_engine(
DATABASE_URL,
pool_pre_ping=True,
pool_size=20,
max_overflow=30,
)
# Session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# TODO: use this get_db_session function in path operation.
def get_db_session(tenant_id: Annotated[str, Header(alias="X-Tenant-ID")]) -> Generator[Session, None, None]:
session = SessionLocal()
try:
# TODO: Implement tenant_id to tenant_schema here
session.execute(text(f"SET search_path TO {tenant_id};"))
session.commit() # Ensure the schema change is applied immediately
yield session
finally:
session.close()