Короткий ответ
Нужно выбрать модель изоляции данных (отдельная БД/схема на tenant против общей таблицы с колонкой tenant_id и обязательной фильтрацией на каждом запросе) и способ определения tenant на входе запроса (поддомен, заголовок, JWT claim), причём определение tenant должно происходить как можно раньше в стеке middleware и передаваться явно во все последующие слои, включая запросы к БД.
Как это работает подробнее
Схема «общая таблица + tenant_id» проще в эксплуатации (одна БД, одни миграции) и дешевле для большого числа мелких tenant, но требует железной дисциплины — каждый запрос к БД должен фильтроваться по tenant_id (желательно на уровне, который невозможно случайно забыть, например через обязательный параметр в репозитории или row-level security в PostgreSQL), иначе утечка данных между tenant — вопрос времени. Схема «отдельная БД/схема на tenant» даёт более сильную изоляцию и проще для выполнения требований compliance отдельных крупных клиентов, но усложняет миграции (нужно прогонять на каждую БД/схему) и операционную нагрузку с ростом числа tenant.
Общая схема с tenant_id дешевле в эксплуатации и миграциях при большом числе tenant, но риск утечки данных между клиентами выше при человеческой ошибке в коде, если не подкреплена защитой на уровне БД (row-level security) или обязательными обёртками репозиториев. Отдельная БД/схема на tenant даёт сильную изоляцию и проще для контрактов с крупными enterprise-клиентами, но усложняет операционную эксплуатацию (миграции, бэкапы, мониторинг) при росте числа tenant. Выбор зависит от количества и размера tenant, требований к изоляции по контракту/compliance и операционной зрелости команды.
Пример кода
# Middleware определяет tenant как можно раньше и кладёт в контекст
import contextvars
tenant_id_var = contextvars.ContextVar("tenant_id")
@app.middleware("http")
async def resolve_tenant(request, call_next):
subdomain = request.url.hostname.split(".")[0]
tenant = get_tenant_by_subdomain(subdomain)
if tenant is None:
return JSONResponse(status_code=404, content={"error": "unknown_tenant"})
tenant_id_var.set(tenant.id)
return await call_next(request)
# Репозиторий ВСЕГДА требует tenant_id явным параметром, а не берёт "откуда-то":
def get_orders(db, tenant_id: int):
return db.query(Order).filter(Order.tenant_id == tenant_id).all()Что использовать на практике
- Практика. Middleware, определяющий tenant по поддомену/заголовку, должен быть одним из первых в стеке и должен безопасно отказывать (а не угадывать tenant по умолчанию), если tenant не определён; дальше tenant_id должен явно передаваться в слой доступа к данным (через contextvar/параметр), а не выводиться неявно где-то в глубине кода.
Подводные камни
- Риск. Полагаются только на фильтрацию на уровне ORM-запросов в бизнес-коде без дополнительной защиты (например, row-level security в БД), и один забытый
WHERE tenant_id = ...в новом запросе становится утечкой данных между клиентами; выбирают общую таблицу для всех tenant без проверки, что индексы и планы запросов продолжают быть эффективными, когда количество строк на одного крупного tenant становится значительно больше, чем у остальных.