python - 如何将来自两个不同来源(没有 RDBMS 关系)的数据合并到单个序列化程序中？

Question

我正在尝试序列化一些对象，这些对象的数据存储在 2 个数据库中，由通用 UUID 链接。第二个数据库 DB2 存储个人数据，因此它作为隔离的微服务运行以遵守各种隐私法。我收到的数据是经过解码的字典列表(而不是模型实例的实际查询集)。我如何调整 ModelSerializer 以序列化此数据？

这是与 DB2 交互以获取个人数据的最小示例:

# returns a list of dict objects, approx representing PersonalData.__dict__
# `custom_filter` is a wrapper for the Microservice API using `requests`
personal_data = Microservice.objects.custom_filter(uuid__in=uuids)

下面是序列化它的最小方法，包括出生日期:

class PersonalDataSerializer(serializers.Serializer):
    uuid = serializers.UUIDField() # common UUID in DB1 and DB2
    dob = serializers.DateField() # personal, so can't be stored in DB1

在我的应用程序中，我需要将 Person 查询集和相关的 personal_data 序列化为一个 JSON 数组。

class PersonSerializer(serializers.ModelSerializer):
    dob = serializers.SerializerMethodField()
    # can't use RelatedField for `dob` because the relationship isn't
    # codified in the RDBMS, due to it being a separate Microservice.

    class Meta:
        model = Person
        # A Person object has `uuid` and `date_joined` fields.
        # The `dob` comes from the personal_data, fetched from the Microservice
        fields = ('uuid', 'date_joined', 'dob',)

    def get_dob(self):
        raise NotImplementedError # for the moment

我不知道是否有一种很好的 DRF 方法可以将两者联系起来。我绝对不想通过在 get_dob 中包含单个请求来向微服务发送(可能有数千个)单个请求。实际 View 看起来像这样:

class PersonList(generics.ListAPIView):
    model = Person
    serializer_class = PersonSerializer

    def get_queryset(self):
        self.kwargs.get('some_filter_criteria')
        return Person.objects.filter(some_filter_criteria)

将微服务数据链接到序列化程序的逻辑应该去哪里，它应该是什么样子？

Answer 1

我建议您覆盖序列化程序和列表方法。

序列化器:

class PersonSerializer(models.Serializer):
    personal_data = serializers.DictField()

    class Meta:
        model = Person

创建一个函数来将 personal_data 字典添加到 persons 对象。在将人员对象列表提供给序列化程序之前使用此方法。

def prepare_persons(persons):
    person_ids = [p.uuid for p in persons]
    personal_data_list = Microservice.objects.custom_filter(uuid__in=person_ids)
    personal_data_dict = {pd['uuid']: pd for pd in personal_data_list}
    for p in persons:
        p.personal_data = personal_data_dict[p.id]
    return persons


def list(self, request, *args, **kwargs):

    queryset = self.filter_queryset(self.get_queryset())

    page = self.paginate_queryset(queryset)

    if page is not None:
        page = prepare_persons(page)
        serializer = self.get_serializer(page, many=True)
        return self.get_paginated_response(serializer.data)
    else:
        persons = prepare_persons(queryset)

    serializer = self.get_serializer(persons, many=True)
    return Response(serializer.data)