action_hero.actions.queue_celery

Provides an admin action to queue Celery tasks for selected records.

class action_hero.actions.queue_celery.QueueCeleryAction

Bases: AdminActionBaseClass

Generates an admin action for queuing a Celery task for a chosen set of records.

Example usage:

conditional_action = QueueCeleryAction(
    task=my_celery_task,
    condition=lambda record: record.should_process(),
)
another_action = QueueCeleryAction(...)

@celery.task
def my_celery_task(record_id):
    record = MyModel.objects.get(pk=record_id)
    ...

class MyModelAdmin(admin.ModelAdmin):
    actions = [conditional_action, QueueCeleryAction(...), another_action]
    model = MyModel
__init__(task, *, condition=None, name=None, short_description=None)

Initializes the action with a Celery task and an optional condition.

Parameters:

  • task (celery.Task) – Should be a Celery Task callable that takes a single model instance’s primary key as an argument.

  • condition (Condition | None) – A callable that takes a model instance and returns a Boolean indicating whether to queue the task for that record.

  • name (str | None) – The action’s name in the admin. If it is omitted, the name of the task will be used instead.

  • short_description (str | None) – The action’s name displayed in the admin. Overrides name.

Return type:

None

handle_item(item)

Queues the Celery task for the given item.

Parameters:

item (Model) – The model instance being processed.