Multi-Tenancy: Build SaaS for Multiple Clients - Data Isolation Strategies

Problem

Building a SaaS platform requires serving multiple independent clients (tenants) from a single application instance while ensuring:

• Data isolation: Tenant A cannot access Tenant B's data
• Security: Unauthorized access prevention
• Scalability: Efficient resource utilization
• Cost efficiency: Shared infrastructure without compromising privacy
• Compliance: Meeting regulatory requirements (GDPR, HIPAA, etc.)

Without proper multi-tenancy architecture, you risk data breaches, performance issues, and regulatory violations.

---

Solution

Multi-Tenancy Models

1. Shared Database, Shared Schema (Row-Level Isolation)

• Single database, single schema
• Tenant identifier in every table
• Most cost-efficient, highest security risk
• Best for: Low-security, high-volume SaaS

2. Shared Database, Separate Schema

• Single database, separate schemas per tenant
• Better isolation than row-level
• Moderate cost and complexity
• Best for: Medium-security requirements

3. Separate Database per Tenant

• Complete database isolation
• Highest security and compliance
• Higher operational overhead
• Best for: Enterprise, high-security requirements

4. Hybrid Approach

• Combination of above strategies
• Shared infrastructure for non-sensitive data
• Isolated databases for sensitive data

---

Code Implementation

1. Row-Level Isolation (Shared Database, Shared Schema)

# Django ORM Example with Row-Level Isolation

from django.db import models
from django.contrib.auth.models import User
from django.http import HttpRequest
from functools import wraps

# Tenant Model
class Tenant(models.Model):
    name = models.CharField(max_length=255)
    slug = models.SlugField(unique=True)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.name

# Base model with tenant isolation
class TenantAwareModel(models.Model):
    tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        abstract = True
        indexes = [
            models.Index(fields=['tenant', 'id']),
        ]

# Example: Customer model
class Customer(TenantAwareModel):
    name = models.CharField(max_length=255)
    email = models.EmailField()
    phone = models.CharField(max_length=20)

    class Meta:
        unique_together = ('tenant', 'email')

    def __str__(self):
        return f"{self.name} ({self.tenant.slug})"

# Example: Order model
class Order(TenantAwareModel):
    customer = models.ForeignKey(Customer, on_delete=models.CASCADE)
    order_number = models.CharField(max_length=50)
    total_amount = models.DecimalField(max_digits=10, decimal_places=2)
    status = models.CharField(
        max_length=20,
        choices=[('pending', 'Pending'), ('completed', 'Completed')]
    )

    class Meta:
        unique_together = ('tenant', 'order_number')

# Middleware to extract tenant from request
class TenantMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # Extract tenant from subdomain or URL
        tenant_slug = self.extract_tenant_slug(request)

        try:
            request.tenant = Tenant.objects.get(slug=tenant_slug)
        except Tenant.DoesNotExist:
            request.tenant = None

        response = self.get_response(request)
        return response

    def extract_tenant_slug(self, request):
        # From subdomain: tenant.example.com
        host = request.get_host().split(':')[0]
        parts = host.split('.')

        if len(parts) > 2:
            return parts[0]

        # Fallback: from URL path
        return request.path.split('/')[1]

# Custom QuerySet for automatic tenant filtering
class TenantAwareQuerySet(models.QuerySet):
    def for_tenant(self, tenant):
        return self.filter(tenant=tenant)

class TenantAwareManager(models.Manager):
    def get_queryset(self):
        return TenantAwareQuerySet(self.model, using=self._db)

    def for_tenant(self, tenant):
        return self.get_queryset().for_tenant(tenant)

# Update models to use custom manager
class Customer(TenantAwareModel):
    name = models.CharField(max_length=255)
    email = models.EmailField()

    objects = TenantAwareManager()

    class Meta:
        unique_together = ('tenant', 'email')

# View with automatic tenant isolation
from django.shortcuts import render
from django.http import HttpResponseForbidden

def customer_list(request):
    if not request.tenant:
        return HttpResponseForbidden("Tenant not found")

    # Automatically filtered by tenant
    customers = Customer.objects.for_tenant(request.tenant)

    return render(request, 'customers/list.html', {
        'customers': customers,
        'tenant': request.tenant
    })

def create_customer(request):
    if request.method == 'POST':
        name = request.POST.get('name')
        email = request.POST.get('email')

        # Automatically associate with current tenant
        customer = Customer.objects.create(
            tenant=request.tenant,
            name=name,
            email=email
        )

        return redirect('customer_detail', pk=customer.id)

    return render(request, 'customers/form.html')

# Decorator for tenant verification
def tenant_required(view_func):
    @wraps(view_func)
    def wrapper(request, *args, **kwargs):
        if not request.tenant:
            return HttpResponseForbidden("Tenant not found")
        return view_func(request, *args, **kwargs)
    return wrapper

@tenant_required
def customer_detail(request, pk):
    try:
        customer = Customer.objects.get(
            id=pk,
            tenant=request.tenant
        )
    except Customer.DoesNotExist:
        return HttpResponseForbidden("Access denied")

    return render(request, 'customers/detail.html', {
        'customer': customer
    })

2. Schema-Level Isolation (Separate Schema per Tenant)

# PostgreSQL with separate schemas

from django.db import connection
from django.core.management.base import BaseCommand

class TenantSchemaManager:
    @staticmethod
    def create_schema(tenant_slug):
        """Create a new schema for tenant"""
        with connection.cursor() as cursor:
            cursor.execute(f"CREATE SCHEMA IF NOT EXISTS {tenant_slug}")
            connection.commit()

    @staticmethod
    def drop_schema(tenant_slug):
        """Drop tenant schema"""
        with connection.cursor() as cursor:
            cursor.execute(f"DROP SCHEMA IF EXISTS {tenant_slug} CASCADE")
            connection.commit()

    @staticmethod
    def migrate_schema(tenant_slug):
        """Run migrations for specific schema"""
        from django.core.management import call_command

        with connection.cursor() as cursor:
            cursor.execute(f"SET search_path TO {tenant_slug}")

        call_command('migrate', verbosity=0)

# Middleware for schema routing
class SchemaMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        tenant_slug = self.extract_tenant_slug(request)

        try:
            tenant = Tenant.objects.get(slug=tenant_slug)
            request.tenant = tenant

            # Set PostgreSQL search_path to tenant schema
            with connection.cursor() as cursor:
                cursor.execute(f"SET search_path TO {tenant_slug}")

        except Tenant.DoesNotExist:
            request.tenant = None

        response = self.get_response(request)
        return response

    def extract_tenant_slug(self, request):
        host = request.get_host().split(':')[0]
        parts = host.split('.')
        return parts[0] if len(parts) > 2 else request.path.split('/')[1]

# Management command to create tenant
class Command(BaseCommand):
    help = 'Create a new tenant'

    def add_arguments(self, parser):
        parser.add_argument('name', type=str)
        parser.add_argument('slug', type=str)

    def handle(self, *args, **options):
        name = options['name']
        slug = options['slug']

        # Create tenant record
        tenant = Tenant.objects.create(name=name, slug=slug)

        # Create schema
        TenantSchemaManager.create_schema(slug)

        # Run migrations
        TenantSchemaManager.migrate_schema(slug)

        self.stdout.write(
            self.style.SUCCESS(f'Tenant "{name}" created successfully')
        )

3. Database-Level Isolation (Separate Database per Tenant)

# Multi-database routing

from django.conf import settings

class TenantRouter:
    """Route database operations to tenant-specific database"""

    def db_for_read(self, model, **hints):
        if 'tenant' in hints:
            return f"tenant_{hints['tenant'].slug}"
        return 'default'

    def db_for_write(self, model, **hints):
        if 'tenant' in hints:
            return f"tenant_{hints['tenant'].slug}"
        return 'default'

    def allow_relation(self, obj1, obj2, **hints):
        return True

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        return True

# Settings configuration
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'saas_main',
        'USER': 'postgres',
        'PASSWORD': 'password',
        'HOST': 'localhost',
    },
    'tenant_acme': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'tenant_acme_db',
        'USER': 'postgres',
        'PASSWORD': 'password',
        'HOST': 'localhost',
    },
    'tenant_globex': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'tenant_globex_db',
        'USER': 'postgres',
        'PASSWORD': 'password',
        'HOST': 'localhost',
    }
}

DATABASE_ROUTERS = ['myapp.routers.TenantRouter']

# View with database routing
def customer_list(request):
    customers = Customer.objects.using(
        f"tenant_{request.tenant.slug}"
    ).all()

    return render(request, 'customers/list.html', {
        'customers': customers
    })

# Tenant provisioning
class TenantProvisioner:
    @staticmethod
    def provision_tenant(tenant_name, tenant_slug):
        """Create new tenant with dedicated database"""
        from django.core.management import call_command

        # Create database
        db_name = f"tenant_{tenant_slug}_db"
        TenantProvisioner.create_database(db_name)

        # Add to DATABASES
        settings.DATABASES[f"tenant_{tenant_slug}"] = {
            'ENGINE': 'django.db.backends.postgresql',
            'NAME': db_name,
            'USER': 'postgres',
            'PASSWORD': 'password',
            'HOST': 'localhost',
        }

        # Run migrations
        call_command('migrate', database=f"tenant_{tenant_slug}")

        # Create tenant record
        Tenant.objects.create(name=tenant_name, slug=tenant_slug)

    @staticmethod
    def create_database(db_name):
        """Create PostgreSQL database"""
        from django.db import connection

        with connection.cursor() as cursor:
            cursor.execute(f"CREATE DATABASE {db_name}")

4. API Authentication with Tenant Isolation

# REST API with tenant-aware authentication

from rest_framework import serializers, viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.authentication import TokenAuthentication
from rest_framework.permissions import IsAuthenticated

class TenantUser(models.Model):
    """User associated with specific tenant"""
    user = models.OneToOneField(User, on_delete=models.CASCADE)
    tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
    role = models.CharField(
        max_length=20,
        choices=[('admin', 'Admin'), ('user', 'User')]
    )

class TenantAuthentication(TokenAuthentication):
    """Custom authentication that validates tenant"""

    def authenticate(self, request):
        auth = super().authenticate(request)

        if auth is None:
            return None

        user, token = auth

        try:
            tenant_user = TenantUser.objects.get(user=user)
            request.tenant = tenant_user.tenant
            request.tenant_user = tenant_user
        except TenantUser.DoesNotExist:
            return None

        return (user, token)

class IsTenantUser(IsAuthenticated):
    """Permission to verify tenant access"""

    def has_permission(self, request, view):
        if not super().has_permission(request, view):
            return False

        return hasattr(request, 'tenant') and request.tenant is not None

class CustomerSerializer(serializers.ModelSerializer):
    class Meta:
        model = Customer
        fields = ['id', 'name', 'email', 'phone']

class CustomerViewSet(viewsets.ModelViewSet):
    serializer_class = CustomerSerializer
    authentication_classes = [TenantAuthentication]
    permission_classes = [IsTenantUser]

    def get_queryset(self):
        """Filter by current tenant"""
        return Customer.objects.filter(tenant=self.request.tenant)

    def perform_create(self, serializer):
        """Automatically set tenant"""
        serializer.save(tenant=self.request.tenant)

    @action(detail=False, methods=['get'])
    def by_email(self, request):
        """Get customer by email"""
        email = request.query_params.get('email')

        try:
            customer = Customer.objects.get(
                tenant=request.tenant,
                email=email
            )
            serializer = self.get_serializer(customer)
            return Response(serializer.data)
        except Customer.DoesNotExist:
            return Response(
                {'error': 'Customer not found'},
                status=status.HTTP_404_NOT_FOUND
            )

5. Data Encryption for Additional Security

# Field-level encryption

from cryptography.fernet import Fernet
from django.conf import settings
import base64

class EncryptedField(models.CharField):
    """Encrypted database field"""

    def __init__(self, *args, **kwargs):
        self.cipher_suite = Fernet(settings.ENCRYPTION_KEY)
        super().__init__(*args, **kwargs)

    def get_prep_value(self, value):
        if value is None:
            return value

        encrypted = self.cipher_suite.encrypt(value.encode())
        return base64.b64encode(encrypted).decode()

    def from_db_value(self, value, expression, connection):
        if value is None:
            return value

        decrypted = self.cipher_suite.decrypt(
            base64.b64decode(value.encode())
        )
        return decrypted.decode()

# Usage
class Customer(TenantAwareModel):
    name = models.CharField(max_length=255)
    email = EncryptedField(max_length=255)  # Encrypted
    phone = EncryptedField(max_length=20)   # Encrypted

    objects = TenantAwareManager()

# Settings
import os
from cryptography.fernet import Fernet

ENCRYPTION_KEY = Fernet.generate_key()  # Store securely in environment

---

Tips & Best Practices

1. Choose the Right Model

Row-Level:      Cost ↑↑↑  Security ↓    Complexity ↓
Schema-Level:   Cost ↑↑   Security ↑↑   Complexity ↑
Database-Level: Cost ↑    Security ↑↑↑  Complexity ↑↑↑

2. Always Filter by Tenant

• Never query without tenant filter
• Use custom managers/querysets
• Implement at ORM level, not application level

3. Audit Logging

class AuditLog(models.Model):
    tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
    user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
    action = models.CharField(max_length=50)
    model_name = models.CharField(max_length=100)
    object_id = models.IntegerField()
    changes = models.JSONField()
    timestamp = models.DateTimeField(auto_now_add=True)

4. Rate Limiting per Tenant

from django_ratelimit.decorators import ratelimit

@ratelimit(key='request.tenant.slug', rate='100/h')
def api_endpoint(request):
    pass

5. Backup Strategy

• Row-level: Single backup, restore specific tenant data
• Schema-level: Per-schema backups
• Database-level: Dedicated backups per tenant

6. Testing

class TenantTestCase(TestCase):
    def setUp(self):
        self.tenant1 = Tenant.objects.create(name='Tenant 1', slug='t1')
        self.tenant2 = Tenant.objects.create(name='Tenant 2', slug='t2')

    def test_data_isolation(self):
        Customer.objects.create(
            tenant=self.tenant1,
            name='Customer A',
            email='a@example.com'
        )

        # Tenant 2 should not see Tenant 1's data
        self.assertEqual(
            Customer.objects.filter(tenant=self.tenant2).count(),
            0
        )

7. Performance Optimization

• Add indexes on tenant + other fields
• Use database-level partitioning
• Cache tenant configuration
• Monitor query performance per tenant

8. Compliance & Security

• Encrypt sensitive fields
• Implement audit logging
• Regular security audits
• GDPR: Easy data export/deletion per tenant
• SOC 2: Document isolation mechanisms

---

Conclusion

Multi-tenancy requires careful planning:

• Start simple (row-level) and scale up
• Automate tenant provisioning
• Test isolation thoroughly
• Monitor for data leaks
• Document your architecture

Choose the model matching your security, cost, and scalability requirements.