Switching Schemas per Request with schema_translate_map

Pass execution_options={"schema_translate_map": {None: tenant_schema}} to session.execute() on every statement to rewrite all unqualified table references to the target PostgreSQL schema at SQLAlchemy compile time — no model changes required. This is the recommended pattern in the Dynamic Schema and Multi-Tenant Routing architecture for SaaS applications that isolate tenants into separate schemas within a shared database.

Quick Answer

The before/after contrast shows the 1.4 workaround (manual string formatting) versus the clean 2.0 approach:

# Legacy pattern — do NOT use (SQL injection risk, breaks compiled caching)
schema = "acme"
raw_sql = f"SELECT * FROM {schema}.invoices WHERE paid = false"
session.execute(text(raw_sql))

# SQLAlchemy 2.0 — schema_translate_map rewrites compiled SQL safely
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from your_models import Invoice  # Invoice.__table_args__ = {"schema": None}

async def get_unpaid_invoices(session: AsyncSession, tenant_schema: str) -> list[Invoice]:
    stmt = select(Invoice).where(Invoice.paid == False)
    result = await session.execute(
        stmt,
        execution_options={"schema_translate_map": {None: tenant_schema}},
    )
    return result.scalars().all()
    # Emits: SELECT invoices.id, ... FROM acme.invoices WHERE invoices.paid = false

The map key (None) matches the schema value declared in the model's __table_args__. If your model declares schema="shared", the key must be "shared", not None.

Execution Context & Async Workflow Integration

Why schema_translate_map is Connection-Pool-Safe

SQLAlchemy compiles each select() or insert() statement into a ClauseElement object and caches the compiled SQL string. When schema_translate_map is applied, it acts as a compilation modifier: the cache key includes the map, so {None: "acme"} and {None: "globex"} produce two distinct cached SQL strings. This means:

1. The rewriting happens entirely in Python, before the SQL reaches the DBAPI.
2. No connection-level session variable is set on the database server.
3. The underlying asyncpg connection is returned to the pool in a clean state after the query, with no tenant-specific residue.

This is fundamentally different from SET search_path = acme (a server-side session variable), which persists on the connection until explicitly reset — a serious problem with PgBouncer or any pooled cloud database deployment that reuses connections across tenants.

Async Session Factory Setup

The async_sessionmaker factory is shared globally, and its async engine and connection pool configuration governs how many concurrent per-tenant requests the process can serve. Schema context is injected per statement, not per session:

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    create_async_engine,
    async_sessionmaker,
)

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/saas_db",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
)

# expire_on_commit=False is critical in async FastAPI: after commit the response
# serializer still needs to read attributes, and the session may be closed.
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

FastAPI Dependency Wiring

The tenant schema should be derived from a verified claim — a JWT sub-tenant field, a validated subdomain, or a server-side lookup — never passed raw from a request header. The FastAPI and Starlette async integration patterns scope the session to the request so the map is applied within a clean transaction boundary.

from fastapi import Depends, FastAPI, Header, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from typing import AsyncGenerator, Annotated

app = FastAPI()

# Allowlist of valid schema names — populated from your tenants table at startup
VALID_SCHEMAS: dict[str, str] = {
    "acme": "acme",
    "globex": "globex",
    "initech": "initech",
}


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session


def resolve_tenant_schema(
    x_tenant_id: Annotated[str, Header()],
) -> str:
    schema = VALID_SCHEMAS.get(x_tenant_id)
    if schema is None:
        raise HTTPException(status_code=403, detail="Unknown tenant")
    return schema


@app.get("/invoices")
async def list_invoices(
    schema: str = Depends(resolve_tenant_schema),
    session: AsyncSession = Depends(get_db),
):
    from your_models import Invoice

    stmt = select(Invoice).where(Invoice.paid == False).order_by(Invoice.id)
    result = await session.execute(
        stmt,
        execution_options={"schema_translate_map": {None: schema}},
    )
    invoices = result.scalars().all()
    return [{"id": inv.id, "amount": inv.amount_cents} for inv in invoices]

Applying the Map to Write Operations

schema_translate_map works identically for insert(), update(), and delete() statements compiled through SQLAlchemy Core:

from sqlalchemy import insert, update
from sqlalchemy.ext.asyncio import AsyncSession
from your_models import Invoice
import datetime


async def create_invoice(
    session: AsyncSession,
    tenant_schema: str,
    amount_cents: int,
) -> int:
    stmt = (
        insert(Invoice)
        .values(tenant_id=1, amount_cents=amount_cents, paid=False)
        .returning(Invoice.id)
    )
    result = await session.execute(
        stmt,
        execution_options={"schema_translate_map": {None: tenant_schema}},
    )
    await session.commit()
    return result.scalar_one()
    # Emits: INSERT INTO acme.invoices (tenant_id, amount_cents, paid)
    #        VALUES (1, 5000, false) RETURNING invoices.id

Models with Multiple Declared Schemas

When your application mixes unschema'd models (public tables) with per-tenant models, use a multi-entry map:

from your_models import Invoice, Product, GlobalConfig

# GlobalConfig lives in the public schema — leave it unmapped
# Invoice and Product live in None schema — map to tenant
stmt = select(Invoice, Product).join(Product, Invoice.product_id == Product.id)
result = await session.execute(
    stmt,
    execution_options={"schema_translate_map": {None: "acme", "public": "public"}},
)

Including "public": "public" in the map is a no-op but makes intent explicit and prevents accidental rewriting if models are later moved.

Using schema_translate_map Inside Background Tasks

FastAPI background tasks and Celery workers need the same per-operation schema routing as request handlers. The key difference is that there is no incoming HTTP request to derive the tenant from — the schema name must be passed explicitly as a parameter:

from fastapi import BackgroundTasks
from sqlalchemy.ext.asyncio import AsyncSession


async def send_invoice_email_background(session: AsyncSession, schema: str, invoice_id: int):
    """Background task: fetch invoice from tenant schema and send email."""
    from sqlalchemy import select
    from your_models import Invoice

    stmt = select(Invoice).where(Invoice.id == invoice_id)
    result = await session.execute(
        stmt,
        execution_options={"schema_translate_map": {None: schema}},
    )
    invoice = result.scalar_one_or_none()
    if invoice:
        # ... dispatch email via SMTP or SES
        pass


@app.post("/invoices/{invoice_id}/send")
async def trigger_invoice_email(
    invoice_id: int,
    background_tasks: BackgroundTasks,
    schema: str = Depends(resolve_tenant_schema),
    session: AsyncSession = Depends(get_db),
):
    background_tasks.add_task(
        send_invoice_email_background, session, schema, invoice_id
    )
    return {"queued": True}

Note that the session passed to a BackgroundTasks callback must remain open for the duration of the background task. Because the AsyncSession lifecycle governs when the connection returns to the pool, using async with AsyncSessionLocal() as session inside the background function (rather than injecting the request-scoped session) is safer for long-running tasks — it prevents holding the session open past response delivery.

Resolving Warnings, Errors & Common Mistakes

Advanced schema_translate_map Optimization

Middleware-Level Schema Injection

Rather than threading tenant_schema through every function call, a Starlette middleware can attach the resolved schema to the request state and a session factory wrapper can read it automatically:

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request


class TenantSchemaMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        tenant_id = request.headers.get("X-Tenant-ID", "")
        schema = VALID_SCHEMAS.get(tenant_id)
        if schema is None and request.url.path.startswith("/api/"):
            from starlette.responses import JSONResponse
            return JSONResponse({"detail": "Unknown tenant"}, status_code=403)
        request.state.tenant_schema = schema
        return await call_next(request)


async def get_schema(request: Request) -> str:
    return request.state.tenant_schema


async def get_tenant_db(
    session: AsyncSession = Depends(get_db),
    schema: str = Depends(get_schema),
):
    """Yields a session with tenant schema pre-bound in session.info."""
    session.info["schema"] = schema
    yield session

All route handlers that depend on get_tenant_db can then call tenant_exec_opts(session) without caring where the schema name came from. This eliminates duplicated schema-resolution logic across dozens of endpoints.

Caching the Resolved Schema in session.info

For endpoints that call multiple ORM helpers, resolving and passing tenant_schema through every function signature is tedious. Attach it once per session and let helpers read it:

from sqlalchemy.ext.asyncio import AsyncSession


async def get_tenant_session(tenant_id: str) -> AsyncSession:
    schema = VALID_SCHEMAS[tenant_id]
    session = AsyncSessionLocal()
    session.info["schema"] = schema
    return session


def tenant_exec_opts(session: AsyncSession) -> dict:
    return {"schema_translate_map": {None: session.info["schema"]}}


# In any repository function:
async def count_invoices(session: AsyncSession) -> int:
    from sqlalchemy import func, select
    from your_models import Invoice

    stmt = select(func.count()).select_from(Invoice)
    result = await session.execute(stmt, execution_options=tenant_exec_opts(session))
    return result.scalar_one()

This pattern eliminates the risk of passing the wrong schema to an inner function and makes tenant context inspectable via session.info in logging middleware.

Combining schema_translate_map with selectinload

Eager-loaded relationships also respect schema_translate_map when the relationship target model declares a matching schema. The map is applied to the secondary SELECT ... IN (...) query that selectinload emits:

from sqlalchemy.orm import selectinload
from your_models import Tenant, Invoice


async def fetch_tenant_with_invoices(
    session: AsyncSession, tenant_id: int, schema: str
) -> Tenant:
    stmt = (
        select(Tenant)
        .where(Tenant.id == tenant_id)
        .options(selectinload(Tenant.invoices))
    )
    result = await session.execute(
        stmt,
        execution_options={"schema_translate_map": {None: schema}},
    )
    return result.scalar_one()
    # Both SELECT tenants... and SELECT invoices WHERE tenant_id IN (...)
    # are rewritten to the tenant schema.

Validating Schema Names Before Use

The single most important security control in schema-per-tenant routing is schema name validation. A schema name in PostgreSQL is an identifier, not a parameter — it cannot be safely passed through a %s placeholder. Always validate against one of these strategies:

1. Database allowlist: Query SELECT schema_name FROM information_schema.schemata WHERE schema_name = $1 before using the name. This is authoritative but adds a round-trip.
2. In-memory allowlist: Build a set[str] of valid schema names at application startup from the tenants table and refresh it periodically (every 60 seconds) via a background task.
3. Regex whitelist: re.fullmatch(r"[a-z][a-z0-9_]{0,62}", schema_name) rejects all SQL injection payloads while allowing valid PostgreSQL identifiers. Use this as a secondary guard, not a substitute for the database allowlist.

import re
from functools import lru_cache


_VALID_SCHEMA_PATTERN = re.compile(r"^[a-z][a-z0-9_]{0,62}$")
_schema_allowlist: set[str] = set()


def validate_schema(schema: str) -> str:
    if not _VALID_SCHEMA_PATTERN.match(schema):
        raise ValueError(f"Invalid schema name format: {schema!r}")
    if schema not in _schema_allowlist:
        raise LookupError(f"Unknown tenant schema: {schema!r}")
    return schema

Refresh _schema_allowlist from a background asyncio task that runs every 60 seconds:

import asyncio
from sqlalchemy import text


async def refresh_schema_allowlist(engine) -> None:
    while True:
        async with engine.connect() as conn:
            result = await conn.execute(text("SELECT schema_name FROM tenants"))
            _schema_allowlist.clear()
            _schema_allowlist.update(row[0] for row in result)
        await asyncio.sleep(60)

Frequently Asked Questions

Related

• Dynamic Schema and Multi-Tenant Routing — Parent guide covering schema isolation models, engine-per-tenant patterns, and vertical partitioning.
• Routing Reads to Replicas with Async Engines — Combine replica routing with schema routing for full production multi-tenant architecture.
• Async Engines, Dialects and Connection Pooling — Connection pool configuration that supports concurrent per-tenant requests safely.
• Integrating SQLAlchemy Async with FastAPI and Starlette — Dependency injection patterns for session lifecycle management in FastAPI.