Django风格合并数据库与API数据至代理模型及CRUD实现方案咨询

下面是符合Django/DRF风格的完整实现思路，涵盖数据查询、API获取、数据合并及CRUD操作：

1. 分层架构设计 #

整体分为三层：

• 数据层：Django ORM模型（本地数据）+ 独立API客户端（外部数据）
• 代理模型层：用DRF Serializer作为数据合并载体，同时承担数据验证和序列化职责
• 视图层：基于DRF ViewSet实现CRUD，协调本地模型和外部API的交互

2. 查询数据库数据 #

直接用Django ORM处理本地数据，定义常规模型即可：

# models.py
from django.db import models

class Product(models.Model):
    sku = models.CharField(max_length=50, unique=True, verbose_name="商品SKU")
    name = models.CharField(max_length=200, verbose_name="商品名称")
    description = models.TextField(null=True, blank=True, verbose_name="商品描述")
    created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")

    def __str__(self):
        return self.name

查询示例：

# 单条查询
product = Product.objects.get(sku="PROD001")
# 批量查询
products = Product.objects.filter(name__contains="手机")

3. 获取外部API数据 #

封装独立的API客户端类，和业务逻辑解耦，方便维护和测试：

# clients.py
import requests
from django.conf import settings
from requests.exceptions import RequestException

class ExternalInventoryClient:
    """外部库存API客户端"""
    BASE_URL = settings.EXTERNAL_INVENTORY_API_URL
    TIMEOUT = 5  # 超时时间

    @classmethod
    def get_inventory(cls, sku):
        """根据SKU获取库存数据"""
        try:
            response = requests.get(
                f"{cls.BASE_URL}/inventory/{sku}/",
                timeout=cls.TIMEOUT
            )
            response.raise_for_status()
            return response.json()
        except RequestException as e:
            # 异常降级处理，返回默认值或抛出自定义异常
            return {"stock_quantity": 0, "warehouse_location": "未知"}

    @classmethod
    def create_inventory(cls, sku, data):
        """创建外部库存记录"""
        try:
            response = requests.post(
                f"{cls.BASE_URL}/inventory/",
                json={"sku": sku, **data},
                timeout=cls.TIMEOUT
            )
            response.raise_for_status()
            return response.json()
        except RequestException as e:
            raise ValueError(f"创建外部库存失败: {str(e)}")

    @classmethod
    def update_inventory(cls, sku, data):
        """更新外部库存数据"""
        try:
            response = requests.put(
                f"{cls.BASE_URL}/inventory/{sku}/",
                json=data,
                timeout=cls.TIMEOUT
            )
            response.raise_for_status()
            return response.json()
        except RequestException as e:
            raise ValueError(f"更新外部库存失败: {str(e)}")

    @classmethod
    def delete_inventory(cls, sku):
        """删除外部库存记录"""
        try:
            response = requests.delete(
                f"{cls.BASE_URL}/inventory/{sku}/",
                timeout=cls.TIMEOUT
            )
            response.raise_for_status()
        except RequestException as e:
            raise ValueError(f"删除外部库存失败: {str(e)}")

4. 合并数据到代理模型 #

用DRF Serializer作为"代理模型"，在序列化过程中合并本地和外部数据：

# serializers.py
from rest_framework import serializers
from .models import Product
from .clients import ExternalInventoryClient

class CombinedProductSerializer(serializers.ModelSerializer):
    # 新增外部API返回的字段
    stock_quantity = serializers.IntegerField(read_only=True, label="库存数量")
    warehouse_location = serializers.CharField(read_only=True, label="仓库位置")

    class Meta:
        model = Product
        fields = ["id", "sku", "name", "description", "created_at", "stock_quantity", "warehouse_location"]

    def to_representation(self, instance):
        """重写序列化方法，合并本地和外部数据"""
        # 先序列化本地模型数据
        rep = super().to_representation(instance)
        # 获取外部库存数据并合并
        inventory_data = ExternalInventoryClient.get_inventory(instance.sku)
        rep.update(inventory_data)
        return rep

5. 实现CRUD操作 #

基于DRF ViewSet实现完整的CRUD，协调本地模型和外部API的操作：

# views.py
from rest_framework import viewsets, status
from rest_framework.response import Response
from django.db import transaction
from .models import Product
from .serializers import CombinedProductSerializer
from .clients import ExternalInventoryClient

class ProductViewSet(viewsets.ModelViewSet):
    queryset = Product.objects.all()
    serializer_class = CombinedProductSerializer

    @transaction.atomic
    def create(self, request, *args, **kwargs):
        """创建商品：先存本地，再同步到外部API"""
        # 拆分本地字段和外部字段
        local_fields = ["sku", "name", "description"]
        local_data = {k: v for k, v in request.data.items() if k in local_fields}
        external_data = {k: v for k, v in request.data.items() if k in ["stock_quantity", "warehouse_location"]}

        # 验证并创建本地模型
        serializer = self.get_serializer(data=local_data)
        serializer.is_valid(raise_exception=True)
        product = serializer.save()

        # 同步到外部API，失败则回滚本地操作
        try:
            ExternalInventoryClient.create_inventory(product.sku, external_data)
        except ValueError as e:
            transaction.set_rollback(True)
            return Response({"error": str(e)}, status=status.HTTP_400_BAD_REQUEST)

        return Response(serializer.data, status=status.HTTP_201_CREATED)

    def update(self, request, *args, **kwargs):
        """更新商品：按需更新本地和外部数据"""
        partial = kwargs.pop('partial', False)
        instance = self.get_object()

        # 拆分字段
        local_fields = ["name", "description"]
        local_data = {k: v for k, v in request.data.items() if k in local_fields}
        external_data = {k: v for k, v in request.data.items() if k in ["stock_quantity", "warehouse_location"]}

        # 更新本地模型
        if local_data:
            serializer = self.get_serializer(instance, data=local_data, partial=partial)
            serializer.is_valid(raise_exception=True)
            serializer.save()

        # 更新外部API
        if external_data:
            try:
                ExternalInventoryClient.update_inventory(instance.sku, external_data)
            except ValueError as e:
                return Response({"error": str(e)}, status=status.HTTP_400_BAD_REQUEST)

        return Response(self.get_serializer(instance).data)

    @transaction.atomic
    def destroy(self, request, *args, **kwargs):
        """删除商品：先删外部API，再删本地数据"""
        instance = self.get_object()
        sku = instance.sku

        # 先删除外部数据，失败则终止
        try:
            ExternalInventoryClient.delete_inventory(sku)
        except ValueError as e:
            return Response({"error": str(e)}, status=status.HTTP_400_BAD_REQUEST)

        # 删除本地数据
        instance.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

关键优化点 #

• 事务原子性：用transaction.atomic()确保本地操作和外部API操作的一致性，失败时回滚
• 异常降级：外部API不可用时返回默认数据，避免整个服务崩溃
• 缓存策略：对外部API数据添加缓存（比如用Django的cache_page或自定义缓存），减少重复请求
• 异步处理：如果API调用耗时较长，可将外部API操作放入Celery异步任务，避免阻塞视图响应
• 测试隔离：单元测试时可mock API客户端，避免依赖外部服务

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