django-q2/django_q/signals.py

import importlib

from django.db.models.signals import post_save
from django.dispatch import Signal, receiver
from django.utils.translation import gettext_lazy as _

from django_q.conf import logger
from django_q.models import Task


@receiver(post_save, sender=Task)
def call_hook(sender, instance, **kwargs):
    if instance.hook:
        f = instance.hook
        if not callable(f):
            try:
                module, func = f.rsplit(".", 1)
                m = importlib.import_module(module)
                f = getattr(m, func)
            except (ValueError, ImportError, AttributeError):
                logger.error(
                        _("malformed return hook '%(hook)s' for [%(name)s]") % {'hook': instance.hook, 'name': instance.name}
                )
                return
        try:
            f(instance)
        except Exception as e:
            logger.error(
                _(
                    "return hook %(hook)s failed on [%(name)s] because %(error)s"
                ) % {'hook': instance.hook, 'name': instance.name, 'error': str(e)}
            )


# args: task
pre_enqueue = Signal()

# args: func, task
pre_execute = Signal()

# args: task
post_execute = Signal()