softtenant 0.1.0

Retrofit multi-tenancy into existing FastAPI + SQLAlchemy applications

softtenant

Retrofit multi-tenancy into existing FastAPI + SQLAlchemy applications — without rewriting your queries.

---

The problem

Adding multi-tenancy to an existing single-tenant app is tedious and risky. The naive approach — add company_id to every table, then manually filter every query — takes weeks and leaks data the moment a developer forgets a WHERE clause:

# Before: developer must remember to filter EVERY query
orders = session.execute(
    select(Order).where(Order.company_id == current_tenant_id)  # forget this once → data leak
).scalars().all()

With dozens of models and hundreds of query sites, missing even one is guaranteed.

The solution

softtenant hooks into SQLAlchemy's event system and injects the WHERE company_id = <tid> clause automatically on every SELECT, UPDATE, and DELETE against a marked model. Developers write normal queries; the library enforces isolation.

# After: no filter needed — softtenant adds it automatically
orders = session.execute(select(Order)).scalars().all()

# If no tenant context is active, the query raises instead of returning wrong data.
# There is no way to accidentally leak cross-tenant data.

---

Installation

pip install softtenant              # core — SQLAlchemy + Alembic only
pip install softtenant[fastapi]     # adds FastAPI dependency + middleware helpers

Requirements: Python 3.11+, SQLAlchemy 2.x, Alembic 1.13+

---

Quick start (5 minutes)

1. Mark your models

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from softtenant import TenantScopedMixin

class Base(DeclarativeBase):
    pass

class Order(TenantScopedMixin, Base):   # <-- only change from a vanilla model
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    item: Mapped[str]
    amount: Mapped[float]
    # company_id is added automatically by TenantScopedMixin

Inheriting TenantScopedMixin adds a nullable company_id: int column (the migration makes it NOT NULL) and registers the model with the query-scoping hook.

2. Install the session hook (once at startup)

from softtenant import install_scoping_hooks
from sqlalchemy.ext.asyncio import AsyncSession

# Sync apps
from sqlalchemy.orm import Session
install_scoping_hooks(Session)

# Async apps
install_scoping_hooks(AsyncSession)

One call, any time before the first query. Safe to call multiple times — idempotent.

3. Wrap requests in a tenant context

from softtenant import tenant_context

with tenant_context(tenant_id=42):
    orders = session.execute(select(Order)).scalars().all()
    # SQL: SELECT * FROM orders WHERE company_id = 42

4. Generate the Alembic migration

softtenant generate-migration \
    --base myapp.models:Base \
    --out migrations/versions/add_multi_tenancy.py

Then run it normally:

alembic upgrade head

---

Core concepts

TenantScopedMixin

from softtenant import TenantScopedMixin

class Invoice(TenantScopedMixin, Base):
    __tablename__ = "invoices"
    id: Mapped[int] = mapped_column(primary_key=True)
    amount: Mapped[float]

The mixin adds:

• company_id: Mapped[int | None] — nullable in the model definition (migration enforces NOT NULL)
• Auto-registration with the query-scoping system

Opting a model out — set __multi_tenant__ = False to exclude a model from filtering:

class AuditLog(Base):
    __tablename__ = "audit_logs"
    __multi_tenant__ = False        # visible to all tenants regardless of context
    id: Mapped[int] = mapped_column(primary_key=True)
    message: Mapped[str]

install_scoping_hooks(session_class)

Registers two SQLAlchemy event listeners on the given Session class:

Event	What it does
do_orm_execute	Injects WHERE company_id = <tid> into every SELECT, UPDATE, and DELETE against a scoped model
before_flush	Stamps company_id on new objects that don't have it set yet

Idempotent — registering twice has no effect (safe to call from multiple test fixtures or multiple startup handlers).

Tenant context

The active tenant ID is stored in a ContextVar, which is:

• Async-safe: each asyncio Task has its own copy — no cross-request leakage
• Thread-safe: same story for threaded WSGI apps

from softtenant import tenant_context, set_current_tenant, get_current_tenant

# Option A — context manager (recommended)
with tenant_context(42):
    ...   # company_id = 42 here

# Option B — imperative (for middleware)
token = set_current_tenant(42)
try:
    ...
finally:
    token.var.reset(token)     # restores previous value

# Read the current tenant
tid = get_current_tenant()     # raises TenantContextNotSetError if unset

Auto-stamp on INSERT

When you add a new object inside a tenant_context, you never need to set company_id:

with tenant_context(42):
    order = Order(item="Widget", amount=99.0)   # no company_id argument
    session.add(order)
    session.flush()
    assert order.company_id == 42               # stamped automatically

The before_flush hook sets it from the ContextVar before the INSERT fires.

Safety guarantee

If code tries to query a scoped model without an active tenant context, softtenant raises TenantContextNotSetError rather than silently returning no rows or all rows:

# No tenant_context() active — this raises, not leaks
session.execute(select(Order))
# → TenantContextNotSetError: No tenant context is set. ...

This makes the failure mode loud and catches missing context during development rather than in production.

---

FastAPI integration

Option A — Dependency injection (recommended)

The dependency approach is opt-in per route, composable with auth, and overridable in tests.

from typing import Annotated
from fastapi import Depends, FastAPI
from starlette.requests import Request
from softtenant.fastapi import make_tenant_dependency

app = FastAPI()

# Define how to resolve a tenant id from a request
async def resolve_tenant(request: Request) -> int:
    raw = request.headers.get("X-Tenant-ID")
    if not raw:
        from fastapi import HTTPException
        raise HTTPException(status_code=401, detail="X-Tenant-ID required")
    return int(raw)

# Create the dependency
require_tenant = make_tenant_dependency(resolver=resolve_tenant)
TenantDep = Annotated[int, Depends(require_tenant)]

@app.get("/orders")
async def list_orders(_: TenantDep, db: DBDep) -> list[OrderRead]:
    result = await db.execute(select(Order))   # WHERE company_id = <tid> injected
    return result.scalars().all()

@app.post("/orders", status_code=201)
async def create_order(body: OrderCreate, _: TenantDep, db: DBDep) -> OrderRead:
    order = Order(item=body.item, amount=body.amount)  # company_id stamped on flush
    db.add(order)
    await db.commit()
    return order

Resolver signature: (Request) -> int | Awaitable[int] — sync or async, both work.

Testing with overrides:

app.dependency_overrides[require_tenant] = lambda: 99  # pin to tenant 99 in tests

Option B — Middleware

Use middleware when you want every request (including 404s, /docs, health checks) to resolve a tenant automatically.

from softtenant.fastapi import TenantMiddleware

app.add_middleware(TenantMiddleware, resolver=resolve_tenant)

The middleware resolves the tenant, calls set_current_tenant, and resets the context after the response. If the resolver raises HTTPException, it is converted to a JSON error response.

Admin routes — cross-tenant access

Use bypass_tenant_scope() for routes that legitimately need all-tenant data:

from softtenant import bypass_tenant_scope

@app.get("/admin/orders")
async def admin_list_orders(db: DBDep) -> list[OrderRead]:
    with bypass_tenant_scope():
        result = await db.execute(select(Order))   # no WHERE company_id filter
        return result.scalars().all()

bypass_tenant_scope() is an explicit escape hatch — it cannot be entered by accident. It restores the previous scoping state when it exits.

---

Migration generator

CLI

softtenant generate-migration \
    --base myapp.models:Base \
    --out migrations/versions/add_multi_tenancy.py

Options:

Flag	Default	Description
--base MODULE:ATTR	(required)	import_path:attribute of your declarative Base
--out PATH	(required)	Output path for the generated migration file
--companies-table	companies	Name for the tenant lookup table
--tenant-id-column	company_id	Name for the FK column added to scoped tables
--default-tenant-name	Default	Name of the seed tenant inserted during upgrade
--batch-size	500	Rows per UPDATE batch during the backfill step

What the generated migration does

1. Creates companies (id, name, created_at) — the tenant lookup table
2. Inserts a default tenant row (id=1, name="Default")
3. For each tenant-scoped table:
   • Adds company_id INTEGER NULL — fast, no table lock
   • Backfills company_id = 1 in batches using keyset pagination (no OFFSET scan)
   • Alters company_id to NOT NULL
   • Creates an index on company_id
   • Adds a FK constraint to companies.id
4. Downgrade reverses all of the above

All schema changes use op.batch_alter_table() so the migration works on SQLite, PostgreSQL, and MySQL without modification.

Programmatic API

from softtenant.migration import generate_migration

generate_migration(
    base=Base,
    output_path="migrations/versions/add_multi_tenancy.py",
    companies_table="companies",
    tenant_id_column="company_id",
    default_tenant_id=1,
    default_tenant_name="Acme Corp",
    batch_size=1000,
)

---

Seeding data in tests

Use bypass_tenant_scope() when writing seed data that needs explicit company_id values (e.g., in fixtures):

from softtenant import bypass_tenant_scope

with bypass_tenant_scope():
    session.add(Order(id=1, item="Widget", amount=100.0, company_id=1))
    session.add(Order(id=2, item="Gadget", amount=200.0, company_id=2))
    session.commit()

---

Nested contexts

tenant_context() nests correctly — the outer context is restored when the inner one exits:

with tenant_context(1):
    orders_t1 = session.execute(select(Order)).scalars().all()  # tenant 1
    with tenant_context(2):
        orders_t2 = session.execute(select(Order)).scalars().all()  # tenant 2
    orders_t1_again = session.execute(select(Order)).scalars().all()  # tenant 1 again

---

Configuration

Override library defaults once at application startup:

from softtenant import configure

configure(
    companies_table="tenants",   # default: "companies"
    tenant_id_column="tenant_id",  # default: "company_id"
    batch_size=1000,             # default: 500 (migration backfill batch size)
)

configure() raises SoftTenantConfigError on unknown keys, catching typos at startup.

---

Error reference

Exception	When raised
TenantContextNotSetError	A SELECT/UPDATE/DELETE is issued against a scoped model with no active tenant context
TenantResolutionError	The tenant resolver returned a non-integer value
SoftTenantConfigError	configure() was called with an unknown configuration key

All exceptions inherit from SoftTenantError.

---

Examples

The examples/ directory contains two runnable demos:

examples/demo.py — pure SQLAlchemy, no server required:

python examples/demo.py

Demonstrates every core feature (SELECT isolation, auto-stamp, bypass, bulk UPDATE/DELETE, nested contexts) against an in-memory SQLite database.

examples/fastapi_app.py — complete async FastAPI application:

pip install uvicorn aiosqlite
python examples/fastapi_app.py
# Docs at http://localhost:8000/docs

Then:

curl -H "X-Tenant-ID: 1" http://localhost:8000/orders
curl -H "X-Tenant-ID: 2" http://localhost:8000/orders
curl http://localhost:8000/admin/orders   # cross-tenant, no header required

---

How it works

softtenant uses two SQLAlchemy session events:

do_orm_execute — fires before every ORM statement executes. For SELECT, softtenant adds a with_loader_criteria option that SQLAlchemy propagates through joins, subqueries, and eager loads automatically. For UPDATE/DELETE, it appends a WHERE company_id = <tid> clause directly to the statement.

before_flush — fires before new objects are written. softtenant iterates session.new, finds objects that are TenantScopedMixin instances without company_id set, and stamps them from the ContextVar.

The active tenant is stored in a ContextVar[int | None], which is per-asyncio-Task and per-thread, making it safe for both async and threaded environments without any synchronization.

---

Supported databases

Database	Status
SQLite	Full support (tested in every CI run)
PostgreSQL	Full support (CI with service container)
MySQL / MariaDB	Full support (CI with service container)

---

Contributing

git clone https://github.com/suhanapthn24/softtenant
cd softtenant
pip install -e ".[dev]"
pytest

To run against all three database backends locally, set the env vars before running pytest:

export POSTGRES_URL="postgresql+psycopg2://user:pass@localhost/softtenant_test"
export MYSQL_URL="mysql+pymysql://user:pass@localhost/softtenant_test"
pytest

---

License

MIT — see LICENSE.