如何在Django中实现post_save信号？地理定位应用代码求助

以下是给你的代码添加post_save信号的具体步骤：

1. 引入Django信号模块 #

在代码顶部补充信号相关导入：

from django.db.models.signals import post_save
from django.dispatch import receiver

2. 编写信号接收器并连接信号 #

在文件末尾添加接收器函数，仅在用户首次创建时触发Celery任务，避免更新用户时重复执行：

@receiver(post_save, sender=User)
def trigger_enrich_user(sender, instance, created, **kwargs):
    # 仅当用户刚创建且存在IP地址时触发任务
    if created and instance.ip_address:
        enrich_user.delay(instance.pk)

• created参数：Django会在对象首次创建时设为True，更新操作时为False，用它避免重复触发任务
• instance.ip_address判断：确保用户有IP地址再调用API，避免无意义的请求

3. 确保信号在Django启动时加载 #

Django需要在启动阶段注册信号，你可以：

• 将信号代码放在users app的signals.py文件中，然后在users/apps.py的ready()方法中导入：

  # users/apps.py
  from django.apps import AppConfig
  
  class UsersConfig(AppConfig):
      default_auto_field = 'django.db.models.BigAutoField'
      name = 'apps.users'
  
      def ready(self):
          import apps.users.signals  # 导入信号模块
  

• 或者直接将信号代码放在当前的任务文件中（只要该文件会被Django启动时加载到）

完整代码示例 #

整合后的代码如下：

from ipaddress import ip_address
from django.contrib.auth import get_user_model
from django.db.models.signals import post_save
from django.dispatch import receiver

from celery import shared_task
from apps.users.abstractapi import AbstractAPI

User = get_user_model()

@shared_task
def enrich_user(user_pk):
    user = User.objects.get(pk=user_pk)
    api = AbstractAPI()

    location_details = api.get_geolocation_details(ip_address=user.ip_address)
    if location_details is not None:
        user.country = location_details.get("country")
        user.country_code = location_details.get("country_code")
        user.country_geoname_id = location_details.details.get("country_geoname_id")
        user.longitude = location_details.get("longitude")
        user.latitude = location_details.get("latitude")
        user.save(update_fields=("country", "country_code", "country_geoname_id", "longitude", "latitude"))

    holiday_details = api.get_holiday_details(
        country_code=user.country_code,
        day=user.date_joined.day,
        month=user.date_joined.month,
        year=user.date_joined.year,
    )

    if holiday_details is not None and any(holiday_details):
        user.joined_on_holiday = True
        user.save(update_fields=("joined_on_holiday",))

@receiver(post_save, sender=User)
def trigger_enrich_user(sender, instance, created, **kwargs):
    if created and instance.ip_address:
        enrich_user.delay(instance.pk)

关键注意事项 #

• 即便使用update_fields保存用户，post_save信号依然会触发，但通过created判断可避免重复执行任务
• 确保Celery服务已启动，否则任务会堆积在队列中无法执行
• 可给enrich_user任务添加重试机制，应对API调用失败的情况

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