Signals¶

Signals emitted during task lifecycle. Connect to these signals to hook into task execution.

API Reference¶

`signals` ¶

Signals emitted during task lifecycle.

This module defines Django signals that are fired at various points during task execution, allowing you to hook into the task lifecycle.

Signals

before_job: Fired before task execution begins. - sender: Task class - task: The Task instance being executed

on_success: Fired when a task completes successfully. - sender: Task class - task: The completed Task instance

on_failure: Fired when a task fails with an exception or timeout. - sender: Task class - task: The failed Task instance - error: The exception that caused the failure (may be None for orphaned tasks)

before_loop: Fired before each iteration of a generator task. - sender: Task class - task: The Task instance - iteration: Current iteration index (0-based)

after_loop: Fired after each iteration of a generator task. - sender: Task class - task: The Task instance - output: The value yielded by the generator - iteration: Current iteration index (0-based)

Example

Connecting to signals::

from django.dispatch import receiver
from django_simple_queue.signals import on_success, on_failure

@receiver(on_success)
def handle_task_success(sender, task, **kwargs):
    print(f"Task {task.id} completed successfully!")

@receiver(on_failure)
def handle_task_failure(sender, task, error, **kwargs):
    print(f"Task {task.id} failed: {error}")

`before_job = django.dispatch.Signal()` `module-attribute` ¶

Signal fired before task execution begins. Provides: task.

`on_success = django.dispatch.Signal()` `module-attribute` ¶

Signal fired when a task completes successfully. Provides: task.

`on_failure = django.dispatch.Signal()` `module-attribute` ¶

Signal fired when a task fails. Provides: task, error.

`before_loop = django.dispatch.Signal()` `module-attribute` ¶

Signal fired before each generator iteration. Provides: task, iteration.

`after_loop = django.dispatch.Signal()` `module-attribute` ¶

Signal fired after each generator iteration. Provides: task, output, iteration.

Signal Summary¶

Signal	When Fired	Arguments
`before_job`	Before task execution	`task`
`on_success`	Task completed successfully	`task`
`on_failure`	Task failed	`task`, `error`
`before_loop`	Before generator iteration	`task`, `iteration`
`after_loop`	After generator iteration	`task`, `output`, `iteration`

Connecting to Signals¶

Using Decorator¶

from django.dispatch import receiver
from django_simple_queue.signals import on_success, on_failure

@receiver(on_success)
def handle_success(sender, task, **kwargs):
    print(f"Task {task.id} completed")

@receiver(on_failure)
def handle_failure(sender, task, error, **kwargs):
    print(f"Task {task.id} failed: {error}")

Manual Connection¶

from django_simple_queue.signals import before_job

def my_handler(sender, task, **kwargs):
    print(f"Starting {task.task}")

before_job.connect(my_handler)

Signal Arguments¶

before_job¶

@receiver(before_job)
def handler(sender, task, **kwargs):
    # sender: Task class
    # task: Task instance about to execute
    pass

on_success¶

@receiver(on_success)
def handler(sender, task, **kwargs):
    # sender: Task class
    # task: Completed Task instance
    pass

on_failure¶

@receiver(on_failure)
def handler(sender, task, error, **kwargs):
    # sender: Task class
    # task: Failed Task instance
    # error: Exception or None (for orphaned/timeout)
    pass

before_loop / after_loop¶

@receiver(before_loop)
def handler(sender, task, iteration, **kwargs):
    # iteration: 0-based index
    pass

@receiver(after_loop)
def handler(sender, task, output, iteration, **kwargs):
    # output: Value yielded by generator
    # iteration: 0-based index
    pass

Loading Signal Handlers¶

Ensure handlers load on Django startup:

# myapp/apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        import myapp.signals  # noqa

See the Using Signals guide for more examples.