Multi-tenancy

Overview

Organizations are isolated by design: models carry organization_id, CRUD and queries take tenant arguments or rely on session context, and list/search paths must never return another org’s rows.

Multi-tenant Database Design

Organization Model

The foundation of multi-tenancy is the Organization model:

organization-model.py
class Organization(Base):
    name = Column(String, nullable=False)
    slug = Column(String, unique=True, nullable=False)
    is_active = Column(Boolean, default=True)
    # Additional fields...

Organization References

Most models include a reference to the organization they belong to:

organization-reference.py
class SomeModel(Base):
    # ...other fields
    organization_id = Column(GUID(), ForeignKey("organization.id"))
    organization = relationship("Organization")

Direct Parameter Passing Architecture

The application uses direct parameter passing for tenant context instead of session variables, providing better performance and security.

Tenant Context Extraction

Tenant context is extracted from authenticated users and passed directly to CRUD operations:

tenant-context.py
def get_tenant_context(current_user: User = Depends(require_current_user_or_token)):
    """Extract tenant context from authenticated user."""
    return current_user.organization_id, current_user.id

CRUD Operations with Tenant Context

All CRUD operations accept organization_id and user_id parameters explicitly:

crud-with-tenant-context.py
def create_entity(
    db: Session,
    entity_data: EntityCreate,
    organization_id: str,
    user_id: str
) -> Entity:
    """Create entity with explicit tenant context."""
    # Auto-populate tenant fields
    populated_data = _auto_populate_tenant_fields(
        Entity, entity_data.dict(), organization_id, user_id
    )

    db_entity = Entity(**populated_data)
    db.add(db_entity)
    db.commit()
    db.refresh(db_entity)
    return db_entity

Query Filtering

Database queries include organization filtering to prevent data leakage:

query-filtering.py
def get_entities(db: Session, organization_id: str) -> List[Entity]:
    """Get entities filtered by organization."""
    return db.query(Entity).filter(
        Entity.organization_id == UUID(organization_id)
    ).all()

Database Session Management

Simple Session Management

Database sessions are managed without tenant setup overhead:

session-management.py
@contextmanager
def get_db() -> Generator[Session, None, None]:
    """Get a simple database session with transparent transaction management."""
    db = SessionLocal()
    try:
        yield db
        # Commit any pending transactions automatically
        if db.in_transaction():
            db.commit()
    except Exception:
        # Rollback on exception
        if db.in_transaction():
            db.rollback()
        raise
    finally:
        # Close the session
        db.close()

FastAPI Dependencies

FastAPI dependencies provide both database sessions and tenant context:

fastapi-dependencies.py
def get_db_with_tenant_context(tenant_context: tuple = Depends(get_tenant_context)):
    """
    FastAPI dependency that provides both a database session and tenant context.

    This eliminates the need for SET LOCAL commands by providing the tenant
    context directly to CRUD operations.

    Returns:
        tuple: (db_session, organization_id, user_id)
    """
    organization_id, user_id = tenant_context

    with get_db() as db:
        yield db, organization_id, user_id

Usage Patterns

For multi-entity operations (recommended):

multi-entity-operations.py
# Use get_db and pass tenant context directly to CRUD operations
def load_initial_data(organization_id: str, user_id: str):
    with get_db() as db:
        # Pass tenant context directly to CRUD operations
        create_statuses(db, initial_data["status"], organization_id=organization_id, user_id=user_id)
        create_behaviors(db, initial_data["behavior"], organization_id=organization_id, user_id=user_id)
        # Automatic commit on success, rollback on exception

For API request handling:

api-request-handling.py
# Standard dependency injection for regular API endpoints
async def get_tests(
    db: Session = Depends(get_db),
    current_user: dict = Depends(require_current_user)
):
    # Tenant context is set by require_current_user dependency
    return crud.get_tests(db)

When to use each approach:

get_db with direct parameters: Multi-entity operations, background tasks, data migrations, initial data loading
Standard dependencies: Regular API endpoints where tenant context is set by authentication middleware

Authentication Integration

The multi-tenancy system integrates with authentication to set tenant context based on the authenticated user:

authentication-integration.py
async def require_current_user(request: Request, db: Session = Depends(get_db)):
    user = request.session.get("user")
    if not user:
        raise HTTPException(status_code=401, detail="Not authenticated")

    # Set tenant context for database operations
    set_tenant(db, user.get("organization_id"), user.get("sub"))

    return user

API Request Flow

Authenticated request → user/org from auth → tenant set on the session (where applicable) → queries respect RLS/org filters → context cleared after the request.

Superuser Access

Superusers can access data across organizations by bypassing the row-level security policies:

bypass-rls.py
def bypass_rls(db: Session):
    """Temporarily bypass row-level security for superuser operations."""
    db.execute(text("SET app.bypass_rls = true"))

This feature is carefully controlled and only available to authenticated superusers.