You can run multiple tenants in a single FastAPI application, each with its own isolated database schema, without needing separate deployments or even separate application instances.
Let’s see this in action. Imagine we have a Tenant model that stores the identifier for each tenant, and we want to switch the database connection to use a specific tenant’s schema.
from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from typing import Generator
# --- Database Setup ---
DATABASE_URL = "postgresql://user:password@host:port/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Tenant model (stores tenant identifiers and their schema names)
class Tenant(Base):
__tablename__ = "tenants"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, unique=True, index=True)
schema_name = Column(String, unique=True, index=True)
# Example model within a tenant's schema
class Item(Base):
__tablename__ = "items" # This will be overridden by the tenant schema
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
description = Column(String, index=True)
# --- Tenant Management ---
# In a real app, you'd have a way to dynamically create schemas and register tenants.
# For this example, we'll pre-populate.
# Base.metadata.create_all(bind=engine) # Create tenant table if it doesn't exist
# Create a dummy tenant for demonstration
# try:
# db = SessionLocal()
# new_tenant = Tenant(name="tenant_a", schema_name="tenant_a_schema")
# db.add(new_tenant)
# db.commit()
# db.refresh(new_tenant)
# print(f"Created tenant: {new_tenant.name} with schema: {new_tenant.schema_name}")
# except Exception as e:
# db.rollback()
# print(f"Tenant already exists or error: {e}")
# finally:
# db.close()
# --- Dynamic Schema Handling ---
def get_tenant_schema(tenant_name: str) -> str:
# In a real app, this would query your tenant table
# For demonstration, hardcoding
if tenant_name == "tenant_a":
return "tenant_a_schema"
elif tenant_name == "tenant_b":
return "tenant_b_schema"
else:
raise ValueError("Tenant not found")
def get_db(tenant_name: str = "tenant_a") -> Generator:
schema = get_tenant_schema(tenant_name)
# Dynamically switch the schema for the current connection
# This uses PostgreSQL's search_path mechanism
db = SessionLocal()
try:
db.execute(f"SET search_path TO {schema}, public;")
yield db
finally:
db.close()
# --- FastAPI App ---
app = FastAPI()
@app.get("/items/")
def read_items(db: SessionLocal = Depends(get_db)):
# The Item model will now be looked up in the 'tenant_a_schema' (or whichever is active)
items = db.query(Item).all()
return [{"id": item.id, "name": item.name, "description": item.description} for item in items]
@app.post("/items/")
def create_item(name: str, description: str, db: SessionLocal = Depends(get_db)):
new_item = Item(name=name, description=description)
db.add(new_item)
db.commit()
db.refresh(new_item)
return {"id": new_item.id, "name": new_item.name, "description": new_item.description}
# --- To run this example ---
# 1. Set up a PostgreSQL database.
# 2. Create the `tenants` table and insert at least one tenant.
# 3. Create schemas named `tenant_a_schema` and `tenant_b_schema`.
# 4. Inside each schema, create the `items` table with the same structure as the Item model.
# Example SQL for tenant_a_schema:
# CREATE SCHEMA tenant_a_schema;
# CREATE TABLE tenant_a_schema.items (
# id SERIAL PRIMARY KEY,
# name VARCHAR,
# description VARCHAR
# );
# 5. Run with `uvicorn your_file_name:app --reload`
# 6. Access endpoints like:
# GET /items/ (will use tenant_a_schema by default)
# POST /items/?name=TestItem&description=A_test_item
# To specify a different tenant, you'd typically pass it in a header or subdomain.
# For this example, modify the `get_db` dependency or add a path parameter.
# Example (if you added a tenant_name path param to get_db):
# GET /items/?tenant_name=tenant_b
The core idea is that the database connection itself can be told which schema to use for subsequent operations. In PostgreSQL, this is managed by the search_path configuration parameter. When a connection is established, you can set its search_path to include your tenant’s schema, followed by public (for shared tables like tenants). This means when you query for Item, PostgreSQL will first look for an items table in the specified tenant schema and only then in the public schema.
The get_db dependency in FastAPI is the perfect place to implement this dynamic switching. It receives the tenant_name (which you’d extract from a request header, subdomain, or JWT), looks up the corresponding schema_name, and then executes SET search_path TO {schema_name}, public; on the database connection before yielding the SQLAlchemy Session. This ensures that all subsequent ORM operations within that request’s scope will target the correct tenant’s data.
To make this work, you need to ensure that your database user has permissions to access all the necessary schemas. When a new tenant is onboarded, you’ll need to:
- Create a new schema in the database (e.g.,
CREATE SCHEMA tenant_b_schema;). - Create the necessary tables within that new schema (e.g.,
CREATE TABLE tenant_b_schema.items (...);). - Register the tenant in your
tenantstable, mapping its identifier to its schema name.
The most surprising thing about this pattern is how little application code needs to change. You can often keep your SQLAlchemy models and Pydantic schemas largely the same, as the isolation happens at the database connection level. The Item model doesn’t need to know it’s operating within tenant_a_schema; the database handles that translation transparently based on the search_path.
When you deploy this, you’ll typically have a single FastAPI application instance running. Incoming requests are routed, and the get_db dependency intelligently selects the correct schema for that request. This is incredibly efficient, allowing you to serve many tenants from a single deployment, reducing operational overhead.
The search_path is evaluated sequentially. If you have tables with the same name in multiple schemas in the search_path, the one in the schema that appears first will be used. This is why SET search_path TO {schema_name}, public; is common: it prioritizes the tenant’s schema for tenant-specific tables and falls back to public for shared resources.
The next challenge you’ll likely face is managing migrations across multiple tenant schemas.