Django多租户应用移除子域名：首次登录验证失败求助

核心原因 #

首次登录时，认证操作发生在你存储db_alias到session之前：用户提交表单后，Django默认认证系统会先去配置的主数据库查找用户，但你的用户实际存储在租户专属数据库中，因此验证失败；第二次登录时session已存有db_alias，中间件提前切换了数据库，认证才能命中正确的租户库，所以验证成功。

具体解决方案 #

1. 重写认证后端，直接从用户名解析租户并切换数据库 #

不要依赖session传递租户标识，而是在认证阶段直接从用户名提取db_alias，临时切换数据库完成用户验证：

# 在你的应用目录下创建backends.py
from django.contrib.auth.backends import ModelBackend
from django.contrib.auth import get_user_model
from django.db import connection

UserModel = get_user_model()

def extract_db_alias_from_username(username):
    # 实现你的用户名解析逻辑，比如从user@client1提取client1
    if '@' not in username:
        raise ValueError("Invalid username format")
    return username.split('@')[-1]

class TenantBackend(ModelBackend):
    def authenticate(self, request, username=None, password=None, **kwargs):
        if not username or not password:
            return None
        
        try:
            db_alias = extract_db_alias_from_username(username)
            pure_username = username.split('@')[0]  # 去掉租户后缀，只保留用户名
        except ValueError:
            return None  # 格式错误，返回None让其他后端处理
        
        # 临时切换到租户数据库验证用户
        original_db = connection.settings_dict['NAME']
        try:
            connection.settings_dict['NAME'] = db_alias
            user = UserModel.objects.get(username=pure_username)
            if user.check_password(password):
                # 验证成功后将db_alias存入session
                if request:
                    request.session['db_alias'] = db_alias
                return user
        except UserModel.DoesNotExist:
            return None
        finally:
            # 恢复默认数据库，避免污染后续请求
            connection.settings_dict['NAME'] = original_db

然后在settings.py中配置这个认证后端：

AUTHENTICATION_BACKENDS = [
    'your_app.backends.TenantBackend',
    # 可选保留默认后端，处理非租户用户
    'django.contrib.auth.backends.ModelBackend',
]

2. 调整中间件，确保全局数据库切换依赖session #

修改你的AppMiddleware，保证每个请求（包括登录后的后续请求）都能根据session切换到正确的租户数据库：

# middleware.py
from django.db import connection
from your_app.backends import extract_db_alias_from_username

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

    def __call__(self, request):
        db_alias = request.session.get('db_alias', 'default')
        original_db = connection.settings_dict['NAME']
        
        # 切换到租户数据库
        connection.settings_dict['NAME'] = db_alias
        
        response = self.get_response(request)
        
        # 请求结束后恢复默认数据库
        connection.settings_dict['NAME'] = original_db
        return response

3. 移除登录视图中提前存session的代码 #

现在认证后端已经在验证成功后存入db_alias，登录视图中之前的提前存session逻辑会导致冲突，直接删除即可。

4. 优化数据库路由（适配线程本地存储） #

由于Django数据库路由不接受request参数，可结合线程本地存储传递租户标识：

# utils.py
import threading

_local = threading.local()

def set_current_db_alias(db_alias):
    _local.db_alias = db_alias

def get_current_db_alias():
    return getattr(_local, 'db_alias', 'default')

修改中间件的__call__方法，同步线程本地存储：

def __call__(self, request):
    db_alias = request.session.get('db_alias', 'default')
    original_db = connection.settings_dict['NAME']
    
    set_current_db_alias(db_alias)
    connection.settings_dict['NAME'] = db_alias
    
    response = self.get_response(request)
    
    set_current_db_alias('default')
    connection.settings_dict['NAME'] = original_db
    return response

最后更新数据库路由：

# routers.py
from your_app.utils import get_current_db_alias

class TenantRouter:
    def db_for_read(self, model, **hints):
        return get_current_db_alias()
    
    def db_for_write(self, model, **hints):
        return get_current_db_alias()
    
    # 根据你的多租户逻辑实现其他路由方法，比如allow_relation等

关键注意事项 #

• 数据库切换后必须恢复默认配置，避免连接池中的连接被污染，导致后续请求出错。
• 认证后端中要做好用户名格式的异常捕获，防止非法输入引发崩溃。
• 若使用数据库连接池（如django-db-connection-pool），需确保切换数据库的操作适配连接池逻辑，必要时调整连接池配置。

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