如何在Django中添加自定义信号，在第三方API请求完成后触发CreateAPIView

针对你的需求，不需要用全局的request_finished信号，而是通过自定义信号来实现「库存修改API执行完成后，触发库存同步逻辑」的精准控制，步骤如下：

1. 定义自定义信号 #

在你的app下新建signals.py文件，创建一个专门用于库存修改成功后的信号：

from django.dispatch import Signal

# 自定义信号：当第三方API成功修改库存后发送
stock_modified_success = Signal()

2. 在库存修改API中发送信号 #

找到你用来调用第三方API修改库存的那个API类（假设名为StockUpdateAPIView），在确认第三方API执行成功后，发送自定义信号：

from rest_framework.views import APIView
from rest_framework.response import Response
from .signals import stock_modified_success
from .services import modify_stock_via_marketplace  # 假设这是你调用第三方API的方法

class StockUpdateAPIView(APIView):
    def post(self, request):
        # 执行第三方API修改库存
        product_id = request.data.get("product_id")
        success = modify_stock_via_marketplace(product_id, request.data.get("new_stock"))
        
        if success:
            # 库存修改成功，发送信号并传递商品ID
            stock_modified_success.send(
                sender=self.__class__, 
                product_id=product_id
            )
            return Response({"status": "库存修改成功"})
        else:
            return Response({"status": "库存修改失败"}, status=400)

3. 抽离库存同步的核心逻辑 #

不要直接触发CreateAPIView的HTTP请求流程，建议把「获取第三方库存并写入数据库」的核心逻辑抽离到单独的服务层（比如services.py），这样视图和信号接收器都能复用：

from .models import Stock
from .utils import get_stock_from_marketplace  # 假设这是你调用第三方API获取库存的方法
from django.utils import timezone

def fetch_and_sync_stock(product_id):
    # 调用第三方API获取最新库存
    latest_stock = get_stock_from_marketplace(product_id)
    # 写入/更新数据库
    Stock.objects.update_or_create(
        product_id=product_id,
        defaults={"quantity": latest_stock["quantity"], "updated_at": timezone.now()}
    )

然后修改你的CreateAPIView，让它调用这个服务方法：

from rest_framework.generics import CreateAPIView
from .serializers import StockSyncSerializer
from .services import fetch_and_sync_stock

class StockSyncCreateAPIView(CreateAPIView):
    serializer_class = StockSyncSerializer

    def perform_create(self, serializer):
        product_id = serializer.validated_data.get("product_id")
        fetch_and_sync_stock(product_id)

4. 编写信号接收器，触发库存同步 #

在signals.py或者views.py中编写接收器，监听自定义信号并执行库存同步：

from django.dispatch import receiver
from .signals import stock_modified_success
from .services import fetch_and_sync_stock

@receiver(stock_modified_success)
def sync_stock_after_modification(sender, **kwargs):
    product_id = kwargs.get("product_id")
    if product_id:
        fetch_and_sync_stock(product_id)

5. 确保信号被注册 #

在你的app的apps.py中，修改ready方法，确保Django启动时加载信号模块：

from django.apps import AppConfig

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app_name'

    def ready(self):
        import your_app_name.signals  # 导入信号模块，完成接收器注册

关键说明 #

• 为什么不用request_finished？因为它是全局信号，所有请求结束都会触发，无法精准控制只有库存修改成功时才执行同步。
• 为什么抽离服务层逻辑？避免直接调用视图的HTTP处理流程（需要构造request对象，冗余且不符合MVC分层思想），同时让代码更易维护和复用。

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