求基于SQL Server的Django多租户应用实现方案（含核心需求）

1. 租户模型与动态数据库创建 #

首先定义租户元数据模型，存储租户标识、对应数据库名等核心信息：

from django.db import models

class Tenant(models.Model):
    id = models.CharField(max_length=50, primary_key=True)
    name = models.CharField(max_length=100)
    db_name = models.CharField(max_length=100, unique=True)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        db_table = 'tenants'

然后编写工具函数，通过SQL Server原生语句动态创建数据库（Django内置的create_database对SQL Server支持有限，直接执行DDL更可靠）：

from django.db import connections

def create_tenant_database(db_name):
    # 使用默认数据库连接执行创建语句，需确保用户有CREATE DATABASE权限
    with connections['default'].cursor() as cursor:
        # 可根据需求添加文件路径等参数
        cursor.execute(f"CREATE DATABASE [{db_name}]")

创建租户时，先保存租户实例，再调用该函数生成对应数据库：

def create_tenant(tenant_id, name):
    db_name = f"tenant_{tenant_id}"
    tenant = Tenant.objects.create(id=tenant_id, name=name, db_name=db_name)
    create_tenant_database(db_name)
    return tenant

2. 自定义数据库路由 #

通过Django的数据库路由机制实现租户级数据隔离，核心是用线程本地存储传递当前租户上下文：

import threading
from django.conf import settings
from .models import Tenant

# 线程本地存储，避免请求间上下文污染
thread_local = threading.local()

class TenantMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # 从请求头/子域名/session提取租户ID，这里以请求头为例
        tenant_id = request.META.get('HTTP_X_TENANT_ID')
        if tenant_id:
            tenant = Tenant.objects.get(id=tenant_id)
            # 动态将租户数据库配置加入settings
            if tenant.db_name not in settings.DATABASES:
                settings.DATABASES[tenant.db_name] = {
                    'ENGINE': 'django.db.backends.sqlserver',
                    'NAME': tenant.db_name,
                    'USER': settings.DATABASES['default']['USER'],
                    'PASSWORD': settings.DATABASES['default']['PASSWORD'],
                    'HOST': settings.DATABASES['default']['HOST'],
                    'PORT': settings.DATABASES['default']['PORT'],
                }
            thread_local.tenant_db = tenant.db_name
        
        response = self.get_response(request)
        
        # 清理线程上下文，避免内存泄漏
        if hasattr(thread_local, 'tenant_db'):
            del thread_local.tenant_db
        
        return response

class TenantDatabaseRouter:
    def db_for_read(self, model, **hints):
        # 优先使用租户数据库，默认返回default（存储租户元数据）
        return getattr(thread_local, 'tenant_db', 'default')

    def db_for_write(self, model, **hints):
        return getattr(thread_local, 'tenant_db', 'default')

    def allow_relation(self, obj1, obj2, **hints):
        # 只允许同租户内的模型关联
        current_db = getattr(thread_local, 'tenant_db', 'default')
        return obj1._state.db == current_db and obj2._state.db == current_db

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        # default数据库仅迁移租户元数据应用，租户数据库迁移业务应用
        if db == 'default':
            return app_label == 'tenant_management'  # 替换为你的租户元数据应用名
        else:
            return app_label != 'tenant_management'

最后在settings.py中配置中间件和路由：

MIDDLEWARE = [
    # ... 其他中间件
    'your_app.middleware.TenantMiddleware',
]

DATABASE_ROUTERS = ['your_app.routers.TenantDatabaseRouter']

3. Celery组件适配 #

Celery异步任务无请求上下文，需手动传递租户ID并切换数据库：

from celery import Task
from django.conf import settings
from .models import Tenant
import threading

thread_local = threading.local()

class TenantTask(Task):
    def __call__(self, *args, **kwargs):
        # 提取租户ID并加载对应数据库配置
        tenant_id = kwargs.pop('tenant_id', None)
        if tenant_id:
            tenant = Tenant.objects.get(id=tenant_id)
            if tenant.db_name not in settings.DATABASES:
                settings.DATABASES[tenant.db_name] = {
                    'ENGINE': 'django.db.backends.sqlserver',
                    'NAME': tenant.db_name,
                    'USER': settings.DATABASES['default']['USER'],
                    'PASSWORD': settings.DATABASES['default']['PASSWORD'],
                    'HOST': settings.DATABASES['default']['HOST'],
                    'PORT': settings.DATABASES['default']['PORT'],
                }
            thread_local.tenant_db = tenant.db_name
        
        try:
            return self.run(*args, **kwargs)
        finally:
            # 清理线程上下文
            if hasattr(thread_local, 'tenant_db'):
                del thread_local.tenant_db

# 基于自定义基类定义任务
@app.task(base=TenantTask)
def process_tenant_data(data):
    # 任务逻辑自动使用租户数据库
    pass

# 调用任务时传递租户ID
process_tenant_data.delay(your_data, tenant_id=current_tenant_id)

4. 批量迁移所有租户数据库 #

编写自定义管理命令，遍历所有租户数据库执行迁移：

from django.core.management.base import BaseCommand
from django.core.management import call_command
from django.conf import settings
from .models import Tenant

class Command(BaseCommand):
    help = 'Apply migrations to all tenant databases'

    def handle(self, *args, **options):
        # 先迁移存储租户元数据的default数据库
        call_command('migrate', database='default')
        
        # 遍历所有租户执行迁移
        for tenant in Tenant.objects.all():
            if tenant.db_name not in settings.DATABASES:
                settings.DATABASES[tenant.db_name] = {
                    'ENGINE': 'django.db.backends.sqlserver',
                    'NAME': tenant.db_name,
                    'USER': settings.DATABASES['default']['USER'],
                    'PASSWORD': settings.DATABASES['default']['PASSWORD'],
                    'HOST': settings.DATABASES['default']['HOST'],
                    'PORT': settings.DATABASES['default']['PORT'],
                }
            self.stdout.write(f"Migrating database: {tenant.db_name}")
            call_command('migrate', database=tenant.db_name, **options)
        
        self.stdout.write(self.style.SUCCESS("All migrations completed"))

执行命令：python manage.py migrate_all_tenants

关键注意事项 #

• SQL Server权限：确保Django使用的数据库账户拥有CREATE DATABASE及访问所有租户数据库的权限。
• 缓存优化：可将租户数据库配置缓存到Redis等工具，避免每次请求都查询租户模型。
• 备份策略：需为每个租户数据库单独制定备份计划，可通过SQL Server的BACKUP DATABASE语句集成到管理命令中。

内容的提问来源于stack exchange，提问作者jeevu94