如何将共享User ID的两个Django模型合并为单个ViewSet

需求背景 #

两个独立Django应用的模型共享来自外部认证服务的user_id，需构建一个统一的ViewSet，对外表现为单个模型：列表返回合并后的字段结构，创建/更新操作需分别同步到两个模型。

现有模型 #

app1/models.py

class App1User(models.Model):
    user_id = models.UUIDField(unique=True)
    app1_field = models.BooleanField()

app2/models.py

class App2User(models.Model):
    user_id = models.UUIDField(unique=True)
    app2_field = models.BooleanField()

目标响应格式 #

[
    {
        "user_id": "...",
        "app1_field": true/false,
        "app2_field": true/false
    },
    ...
]

实现步骤 #

1. 定义合并序列化器 #

自定义序列化器处理字段合并，并重写create和update方法分别操作两个模型：

from rest_framework import serializers
from app1.models import App1User
from app2.models import App2User
import uuid

class MergedUserSerializer(serializers.Serializer):
    user_id = serializers.UUIDField(required=False, default=uuid.uuid4)
    app1_field = serializers.BooleanField()
    app2_field = serializers.BooleanField()

    def create(self, validated_data):
        # 同步创建两个模型实例
        App1User.objects.create(
            user_id=validated_data['user_id'],
            app1_field=validated_data['app1_field']
        )
        App2User.objects.create(
            user_id=validated_data['user_id'],
            app2_field=validated_data['app2_field']
        )
        return validated_data

    def update(self, instance, validated_data):
        # 根据user_id定位两个模型实例并更新
        user_id = validated_data.get('user_id', instance['user_id'])
        
        if 'app1_field' in validated_data:
            App1User.objects.filter(user_id=user_id).update(
                app1_field=validated_data['app1_field']
            )
        if 'app2_field' in validated_data:
            App2User.objects.filter(user_id=user_id).update(
                app2_field=validated_data['app2_field']
            )
        
        # 返回更新后的合并数据
        updated_data = instance.copy()
        updated_data.update(validated_data)
        return updated_data

2. 实现合并ViewSet #

自定义ViewSet处理查询集合并、单实例获取逻辑：

from rest_framework import viewsets
from rest_framework.exceptions import NotFound
from .serializers import MergedUserSerializer
from app1.models import App1User
from app2.models import App2User
from django.db.models import OuterRef, Subquery

class MergedUserViewSet(viewsets.ModelViewSet):
    serializer_class = MergedUserSerializer
    lookup_field = 'user_id'

    def get_queryset(self):
        # 子查询关联App2User的字段到App1User
        app2_subquery = App2User.objects.filter(user_id=OuterRef('user_id')).values('app2_field')[:1]
        app1_merged = App1User.objects.annotate(
            app2_field=Subquery(app2_subquery)
        ).values('user_id', 'app1_field', 'app2_field')

        # 补充仅存在于App2User的记录（按需保留）
        app1_user_ids = App1User.objects.values_list('user_id', flat=True)
        app2_only = App2User.objects.exclude(user_id__in=app1_user_ids).annotate(
            app1_field=None
        ).values('user_id', 'app1_field', 'app2_field')

        # 合并数据集
        return list(app1_merged) + list(app2_only)

    def get_object(self):
        user_id = self.kwargs['user_id']
        app1_user = App1User.objects.filter(user_id=user_id).first()
        app2_user = App2User.objects.filter(user_id=user_id).first()

        if not app1_user and not app2_user:
            raise NotFound(detail="User not found")

        # 构造合并后的单实例数据
        return {
            'user_id': user_id,
            'app1_field': app1_user.app1_field if app1_user else None,
            'app2_field': app2_user.app2_field if app2_user else None
        }

注意事项 #

• 如果业务确保user_id在两个模型中必同时存在，可删除处理"仅单模型存在"的逻辑，简化查询集和实例获取代码
• 若数据集较大，list()转换会影响性能，可考虑使用数据库层面的UNION查询替代内存合并
• 若需要支持字段部分更新，可在序列化器中设置字段为required=False，并在update方法中判断字段是否存在再执行更新

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