SQLAlchemy 2.0 multi-tenant patterns for restaurant management platform. Trigger: When working with SQLAlchemy models, Alembic migrations, database repositories, or any Python file importing from sqlalchemy, models/, repositories/, or alembic/.
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import String, Integer, ForeignKey, func
from datetime import datetime
from typing import Optional
# CORRECT - SQLAlchemy 2.0 style
class Product(Base):
__tablename__ = "products"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
branch_id: Mapped[int] = mapped_column(ForeignKey("branches.id"), index=True)
price_cents: Mapped[int] = mapped_column(Integer)
status: Mapped[str] = mapped_column(String(20), default="active")
description: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
# WRONG - Legacy style. Never use this.
# name = Column(String(100))
class AuditMixin:
created_at: Mapped[datetime] = mapped_column(default=func.now())
updated_at: Mapped[datetime] = mapped_column(default=func.now(), onupdate=func.now())
deleted_at: Mapped[Optional[datetime]] = mapped_column(nullable=True)
created_by: Mapped[Optional[int]] = mapped_column(nullable=True)
updated_by: Mapped[Optional[int]] = mapped_column(nullable=True)
deleted_by: Mapped[Optional[int]] = mapped_column(nullable=True)
is_active: Mapped[bool] = mapped_column(default=True)
Every model inherits AuditMixin. No exceptions.
| Scope | Required Columns | Auto-filter |
|---|---|---|
| Tenant-scoped | tenant_id | WHERE tenant_id = :tid |
| Branch-scoped | tenant_id + branch_id | WHERE tenant_id = :tid AND branch_id = :bid |
| Public | None | Only for auth/health endpoints |
# Tenant-scoped model
class Category(AuditMixin, Base):
__tablename__ = "categories"
id: Mapped[int] = mapped_column(primary_key=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
name: Mapped[str] = mapped_column(String(100))
# Branch-scoped model
class TableSession(AuditMixin, Base):
__tablename__ = "table_sessions"
id: Mapped[int] = mapped_column(primary_key=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
branch_id: Mapped[int] = mapped_column(ForeignKey("branches.id"), index=True)
table_id: Mapped[int] = mapped_column(ForeignKey("tables.id"))
NEVER write a query without tenant context (except explicitly public endpoints like login/health).
class TenantRepository(Generic[T]):
"""Base repo - auto-adds tenant_id to all queries."""
def __init__(self, session: AsyncSession, model: type[T], tenant_id: int):
self.session = session
self.model = model
self.tenant_id = tenant_id
def _base_query(self, include_deleted: bool = False) -> Select:
stmt = select(self.model).where(self.model.tenant_id == self.tenant_id)
if not include_deleted:
stmt = stmt.where(self.model.is_active == True)
return stmt
async def get_by_id(self, id: int, include_deleted: bool = False) -> T | None:
stmt = self._base_query(include_deleted).where(self.model.id == id)
result = await self.session.execute(stmt)
return result.scalar_one_or_none()
class BranchRepository(TenantRepository[T]):
"""Branch-scoped repo - auto-adds tenant_id AND branch_id."""
def __init__(self, session: AsyncSession, model: type[T], tenant_id: int, branch_id: int):
super().__init__(session, model, tenant_id)
self.branch_id = branch_id
def _base_query(self, include_deleted: bool = False) -> Select:
stmt = super()._base_query(include_deleted)
return stmt.where(self.model.branch_id == self.branch_id)
# CORRECT - Soft delete
async def soft_delete(self, id: int, deleted_by: int) -> None:
entity = await self.get_by_id(id)
if entity:
entity.is_active = False
entity.deleted_at = func.now()
entity.deleted_by = deleted_by
await self.session.flush()
# Cascade soft delete: when parent is soft-deleted, children follow
async def soft_delete_with_children(self, parent_id: int, deleted_by: int) -> None:
parent = await self.get_by_id(parent_id)
if parent:
parent.is_active = False
parent.deleted_at = func.now()
parent.deleted_by = deleted_by
# Cascade to children
children_stmt = (
update(ChildModel)
.where(ChildModel.parent_id == parent_id)
.values(is_active=False, deleted_at=func.now(), deleted_by=deleted_by)
)
await self.session.execute(children_stmt)
await self.session.flush()
# WRONG - Never hard delete
# await session.delete(entity)
# CORRECT - Cents avoid floating point issues
price_cents: Mapped[int] = mapped_column(Integer) # 1599 = $15.99
# WRONG - Never use floats/decimals for prices in the DB
# price: Mapped[float] = mapped_column(Float)
# CORRECT - String enum for portability and readability