SQLAlchemy + Postgres Cheatsheet 12 — Schema-Per-Tenant

async def get_tenant_db(tenant: Tenant = Depends(get_tenant)) -> AsyncSession:
    async with AsyncSessionLocal() as session:
        schema = f"tenant_{tenant.slug}"
        await session.execute(text(f"SET search_path TO {schema}, public"))
        yield session

async def create_tenant(tenant_slug: str):
    async with engine.connect() as conn:
        await conn.execute(text(f"CREATE SCHEMA tenant_{tenant_slug}"))
    
    # Run migrations against the new schema
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", DATABASE_URL)
    # ... (multi-tenant env.py iterates or specific arg) ...
    command.upgrade(cfg, "head", arg=f"tenant_{tenant_slug}")

async with engine.begin() as conn:
    await conn.execute(text(f"SET search_path TO {schema}"))
    await conn.run_sync(Base.metadata.create_all)
command.stamp(cfg, "head")

def run_migrations_online():
    schemas = get_all_tenant_schemas()
    
    for schema in schemas:
        connectable = engine_from_config(...)
        with connectable.connect() as conn:
            conn.execute(text(f"SET search_path TO {schema}"))
            context.configure(
                connection=conn,
                target_metadata=target_metadata,
                version_table_schema=schema,        # per-schema version table
            )
            with context.begin_transaction():
                context.run_migrations()
        connectable.dispose()

async def get_admin_db() -> AsyncSession:
    async with AdminSessionLocal() as session:
        # Don't set search_path; explicit schema-qualified queries
        yield session

@app.get("/admin/all-users")
async def all_users(db = Depends(get_admin_db)):
    # ...UNION across tenant schemas...

import concurrent.futures

def migrate_one(schema):
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", DATABASE_URL)
    command.upgrade(cfg, "head", arg=schema)

with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex:
    list(ex.map(migrate_one, schemas))